KBA KBA - 7 months ago 440
Python Question

saving a list of rows to a Hive table in pyspark

I have a pyspark app. I copied a hive table to my hdfs directory, & in python I

sqlContext.sql
a query on this table. Now this variable is a dataframe I call
rows
. I need to randomly shuffle the
rows
, so I had to convert them to a list of rows
rows_list = rows.collect()
. So then I
shuffle(rows_list)
which shuffles the lists in place. I take the amount of random rows I need
x
:

for r in range(x):
allrows2add.append(rows_list[r])

Now I want to save allrows2add as a hive table OR append an existing hive table (whichever is easier to do). The problem is that I can not do this:

all_df = sc.parallelize(allrows2add).toDF()
Cant do this, schema can't be inferred
ValueError: Some of types cannot be determined by the first 100 rows, please try again with sampling


without putting in the whole schema. The schema of
rows
has 117 columns, so I don't want to type them out. Is there a way to extract the schema of
rows
to help me make allrows2add a dataframe or somehow save as a hive table?
I can do
rows.printSchema()
but not sure how to get it into a schema format as a variable to pass
toDF()
without having to parse all of that text

Thanks

Adding for loop info

#Table is a List of Rows from small Hive table I loaded using
#query = "SELECT * FROM Table"
#Table = sqlContext.sql(query).collect()

for i in range(len(Table)):

rows = sqlContext.sql(qry)
val1 = Table[i][0]
val2 = Table[i][1]
count = Table[i][2]
x = 100 - count

#hivetemp is a table that I copied from Hive to my hfs using:
#create external table IF NOT EXISTS hive temp LIKE hivetableIwant2copy LOCATION "/user/name/hiveBackup";
#INSERT OVERWRITE TABLE hivetemp SELECT * FROM hivetableIwant2copy;

query = "SELECT * FROM hivetemp WHERE col1<>\""+val1+"\" AND col2 ==\""+val2+"\" ORDER BY RAND() LIMIT "+str(x)

rows = sqlContext.sql(query)
rows = rows.withColumn("col4", lit(10))
rows = rows.withColumn("col5", lit(some_string))
#writing to parquet is heck slow AND I can't work with pandas due to the library not installed on the server
rows.saveAsParquetFile("rows"+str(i)+".parquet")
#tried this before and heck slow also
#rows_list = rows.collect()
#shuffle(rows_list)

Answer

When the schema can't be inferred, there's usually a reason. toDF is syntactic sugar for the createDataFrame function, which by default only uses the first 100 rows (despite the docs saying it only uses the first row) to determine what the schema should be. To change this, you can increase the sampling ratio to look at a greater percentage of your data:

df = rdd.toDF(sampleRatio=0.2)
# or...
df = sqlContext.createDataFrame(rdd, samplingRatio=0.2)

It's also possible that your random sample happened to only take rows with empty values for some particular columns. If this is the case, you can either create a schema from scratch like so:

from pyspark.sql.types import *
# all DataFrame rows are StructType
# can create a new StructType with combinations of other types
schema = StructType([
    StructField("column_1", StringType(), True),
    StructField("column_2", IntegerType(), True)
    # etc.
])
df = sqlContext.createDataFrame(rdd, schema=schema)

Or, you can get the schema from the previous DataFrame you created by accessing the schema value:

schema = df1.schema
df2 = sqlContext.createDataFrame(rdd, schema=schema)

Note that if your RDD's rows aren't StructType (a.k.a. Row) objects instead of dictionaries or lists, you won't be able to create a data frame from them. If your RDD rows are dictionaries, you can convert them to Row objects like this:

rdd = rdd.map(lambda x: pyspark.sql.Row(**x))
# ** is to unpack the dictionary since the Row constructor
# only takes keyword arguments