python之如何有效地将大型 .tsv 文件上传到 pyspark 中带有拆分列的 Hive 表

我有一个大型(约 1000 万行).tsv 文件，其中包含两列，“id”和“group”。 'Group' 列实际上是某个 id 所属的所有组的列表，因此文件如下所示:

id1     group1,group2 
id2     group2,group3,group4 
id3     group1 
... 

我需要使用 pyspark 将其上传到 Hive 表，但是我想拆分组列，以便一行中只有一个组，因此生成的表如下所示:

id1    group1 
id1    group2 
id2    group2 
id2    group3 
id2    group4 
id3    group1 

我已尝试逐行阅读，仅使用 python split() 拆分列，然后为每一行创建 spark 数据帧并将其与每次迭代合并。我的代码有效，但效率极低，因为处理 1000 行需要 2 分钟。我的代码如下:

fields = [StructField('user_id', StringType(), True),StructField('group_id', StringType(), True)]  
membership_schema = StructType(fields)  
 
result_df = sqlContext.createDataFrame(sc.emptyRDD(), membership_schema) 
 
with open('file.tsv','r') as f: 
    for line in f: 
        parts = line.split() 
        id_part = parts[0] 
        audience_parts = parts[1].split(',') 
        for item in audience_parts: 
            newRow = sqlContext.createDataFrame([(id_part,item)], membership_schema) 
            result_df = result_df.union(newRow) 
df_writer = DataFrameWriter(result_df) 
df_writer.insertInto("my_table_in_hive") 

有没有一种更简单、更有效的方法可以将整个文件上传到表中而无需遍历行？

感谢您的帮助。

请您参考如下方法:

我查看了上面代码的计划，它似乎扫描了很多，也没有为您提供与 spark 的并行性。 您可以使用spark native方法将文件数据读入更多分区，并控制它们在分区之间均匀分布数据。

df = sc.textFile(file_path,10).map(lambda x: x.split()).map(lambda x :(x[0],x[1].split(","))).toDF(['id','group']) 
from pyspark.sql.functions import explode 
newdf = df.withColumn("group", explode(df.group)) 
 
newdf.write.format("orc").option("header", "true").mode("overwrite").saveAsTable('db.yourHivetable') 

此外，您可以增加或减少进入爆炸的分区大小或控制随机分区。

spark.conf.set("spark.sql.files.maxPartitionBytes","30") 
spark.conf.set("spark.sql.shuffle.partitions", "100")