如果列数不同，如何处理从源 spark df 到 hive 表的插入

Posted 2023-04-15

【中文标题】如果列数不同，如何处理从源 spark df 到 hive 表的插入

【英文标题】：How to handle inserts from a source spark df to hive table if the number of columns differ

【发布时间】：2018-04-27 05:41:04

【问题描述】：

我正在尝试将 pyspark 数据框写入配置单元表，但由于我的源 df 只有 5 列而目标有 9 列，因此会导致错误。

此外，由于存在多种情况，因此我不想构建可以解决此问题的手动插入查询。我正在寻找一种更好的自动化方式来处理这个问题，而无需为每个案例手动编写查询。

我想通过获取源 df 和目标表中存在但不在源 df 中的额外列，在 spark 中创建一个新的 df，但它没有按照我的想法工作。

这是我正在处理的代码

#extract cols from src df and tgt df(hive table) 
src_cols = df1.columns
tgt_cols = df2.columns

#get the extra cols (diff)
extra_cols = list(set(tgt_cols) - set(src_cols))
#extra_cols = ['state', 'datetime', 'zipcode', 'type']

#formulate the string to add extra cols
string = ""
for item in extra_cols:
    string += str(".withColumn(\""+item+"\", lit(\"NULL\"))")

这将打印出我可以用于新 df 的所需字符串

#'.withColumn("state", lit(NULL)).withColumn("datetime", lit(NULL)).withColumn("zipcode", lit(NULL)).withColumn("type", lit(NULL))'


new_df = "df1" + string
#'df1.withColumn("state", lit("NULL")).withColumn("datetime", lit("NULL")).withColumn("zipcode", lit("NULL")).withColumn("type", lit("NULL"))'

现在的问题是我无法执行代码df1.withColumn("state", lit("NULL")).withColumn("datetime", lit("NULL")).withColumn("zipcode", lit("NULL")).withColumn("type", lit("NULL"))，因为它是一个字符串

谁能帮助我以更好的方式处理这种情况。

谢谢。

【问题讨论】：

你没有提到你的目标数据库。它是 NoSQL DB 还是 RDBMS？

我不知道如何在python中做到这一点，但使用scala你可以创建一个包含原始列和lit("NULL").alias("state")等的Column对象的列表。然后你可以使用df.select(col_list: _*)。

@PhaniKumarYadavilli 目标是 hive table。它已经在第一行提到了。

如何简单地使用 SQL INSERT...SELECT 查询来指定哪些源列 （来自“临时视图”） 映射到哪些目标列？ SQL 解析器将处理目标列中没有值的空值。你将摆脱所有讨厌的 Scala 样板代码

【参考方案1】：

如果您已将列名差异列表标识为

#extra_cols = ['state', 'datetime', 'zipcode', 'type']

那么你不需要制定字符串来添加额外的列，你可以简单地使用reduce函数将.withColumn应用到列名列表上

import pyspark.sql.functions as f
to_be_written_df = reduce(lambda temp_df, col_name: temp_df.withColumn(col_name, f.lit('NULL')), extra_cols, df1)

应该可以解决你的问题

【讨论】：

还有一件事与这个问题无关，但是你能指导我使用数据框在 hive 中自动增量插入

研究它。如果您没有找到答案，请再问一个问题