我想將流資料從kafka 主題寫入hive 表。
我可以通過閱讀 kafka 主題來創建資料幀，但由于檔案格式不匹配，資料沒有寫入 Hive 表。我已經指定dataframe.format("parquet")并使用stored as parquet.
下面是代碼片段：
def hive_write_batch_data(data, batchId):
data.write.format("parquet").mode("append").saveAsTable(table)
def write_to_hive(data,kafka_sink_name):
global table
table = kafka_sink_name
data.select(col("key"),col("value"),col("offset")) \
.writeStream.foreachBatch(hive_write_batch_data) \
.start().awaitTermination()
if __name__ == '__main__':
kafka_sink_name = sys.argv[1]
kafka_config = {
....
..
}
spark = SparkSession.builder.appName("Test Streaming").enableHiveSupport().getOrCreate()
df = spark.readStream \
.format("kafka") \
.options(**kafka_config) \
.load()
df1 = df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)","offset","timestamp","partition")
write_to_hive(df1,kafka_sink_name)
Hive 表創建為 Parquet：
CREATE TABLE test.kafka_test(
key string,
value string,
offset bigint)
STORED AS PARQUET;
它給了我錯誤：
pyspark.sql.utils.AnalysisException: "The format of the existing table test.kafka_test is `HiveFileFormat`. It doesn\'t match the specified format `ParquetFileFormat`.;"
如何將資料框寫入配置單元表？
uj5u.com熱心網友回復：
我放下了 hive 表，并運行了 Spark-streaming 作業。表是用正確的格式創建的。
轉載請註明出處，本文鏈接：https://www.uj5u.com/qukuanlian/450581.html
標籤：阿帕奇火花 pyspark apache-spark-sql 火花流
上一篇：來自陣列列的Pyspark樣本值
下一篇：用逗號分割字串，同時保留數字部分