python - Pyspark:使用 Zeppelin Notebook 插入 Hive 中的表时出现问题

Question

注意:这是在 Zeppelin Notebook 中编写的。

我一直在尝试复制源表的内容，将 NULL 转换为 0，然后将其插入到目标表中。但是，我遇到了错误:

AnalysisException: u'\nUnsupported language features in query: INSERT INTO TABLE..

在线阅读时，Spark 似乎不支持 INSERT INTO，但是，我一直在阅读您可以使用数据帧并将其附加到表中。这是我的原始代码:

%pyspark
from pyspark.context import SparkContext

df = sqlContext.sql("SELECT * FROM test_target")

dfr=df.collect()
print(dfr)

这是我尝试实现 INSERT INTO/dataframe 方法的尝试。我仍然收到相同的 AnalysisException 错误。

%pyspark
from pyspark.context import SparkContext

sql = sqlContext.sql("INSERT INTO TABLE kenny_target(`user`, `age`) SELECT `user`, COALESCE(`age`, 0L) FROM kenny_source")
frame= sqlContext.createDataFrame(sql).collect()
frame.write.mode("append").saveAsTable("kenny_source_test")

test = sqlContext.sql("SELECT * FROM test_source_test")
dfr=test.collect()
print(dfr)

Answer 1

您提供了以下代码。

%pyspark
from pyspark.context import SparkContext

sql = sqlContext.sql("INSERT INTO TABLE kenny_target(`user`, `age`) SELECT 
`user`, COALESCE(`age`, 0L) FROM kenny_source")
frame= sqlContext.createDataFrame(sql).collect()
frame.write.mode("append").saveAsTable("kenny_source_test")

由于某些原因您无法执行此操作。

1。您正在尝试再次创建 DataFrame。

假设支持INSERT INTO TABLE。当您执行 sqlContext.sql (它 将创建一个 DataFrame)。

sql = sqlContext.sql("INSERT INTO TABLE kenny_target('user', 'age') SELECT 'user', COALESCE('age', 0L) FROM kenny_source")

接下来是这一行。当你运行这个时，spark 会抛出一个错误。

frame = sqlContext.createDataframe(sql)

TypeError: data is already a DataFrame

2。您正在尝试收集()DataFrame并将其写入表中。

你有这行代码。

# Since, you wrote sql = sqlContext.sql('query')
# You could write frame = sql.collect()
frame = sqlContext.createDataFrame(sql).collect()

Note, after you collect a DataFrame (frame in this case), it is no longer a DataFrame. A collect() always collects the DataFrame as List.

最后一行不起作用，因为 frame 是一个列表(正如您对其执行的collect()操作一样)。

frame.write.mode("append").saveAsTable("kenny_source_test")

它抛出一个错误说明

'list' object has no attribute 'write' # Reason: frame is now a list

分辨率:

# SELECT user and age into 'frame' by creating a DataFrame called as 'frame'
frame = sqlContext.sql('''SELECT `user`, COALESCE(`age`, 0L) FROM kenny_source''')

# Write it to the table: kenny_source_test
frame.write.mode("append").saveAsTable("kenny_source_test")