将行列表保存到 pyspark 中的 Hive 表

Question

我有一个 pyspark 应用程序。我将一个 hive 表复制到我的 hdfs 目录中，并且在 python 中我 sqlContext.sql 对该表进行了查询。现在这个变量是一个我称之为rows的数据框。我需要随机打乱rows，所以我必须将它们转换为行列表rows_list = rows.collect()。然后我shuffle(rows_list) 将列表重新排列到位。我取了我需要的随机行数量x：

for r in range(x): allrows2add.append(rows_list[r]) 现在我想将 allrows2add 保存为一个 hive 表或附加一个现有的 hive 表（以更容易做的为准）。问题是我不能这样做：

all_df = sc.parallelize(allrows2add).toDF() 不能这样做，无法推断架构 ValueError: Some of types cannot be determined by the first 100 rows, please try again with sampling

没有放入整个架构。 rows 的架构有 117 列，所以我不想把它们打出来。有没有办法提取rows 的架构来帮助我制作 allrows2add 数据框或以某种方式保存为配置单元表？ 我可以 rows.printSchema() 但不确定如何将其转换为模式格式作为变量传递 toDF() 而无需解析所有文本

谢谢

添加循环信息

#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 1

当无法推断架构时，通常是有原因的。 toDF 是 createDataFrame 函数的语法糖，默认情况下它只使用前 100 行（despite the docs 说它只使用第一行）来确定模式应该是什么。要改变这一点，您可以提高采样率以查看更大百分比的数据：

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

您的随机样本也有可能碰巧只取某些特定列的空值行。如果是这种情况，您可以像这样create a schema from scratch：

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

或者，您可以通过访问 schema 值从之前创建的 DataFrame 中获取架构：

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

请注意，如果您的 RDD 的行不是 StructType（又名Row）对象而不是字典或列表，您将无法从它们创建数据框。如果您的 RDD 行是字典，您可以将它们转换为 Row 对象，如下所示：

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