使用 pyspark 将数据从 kafka 写入 hive - 卡住

Posted 2023-04-15

【中文标题】使用 pyspark 将数据从 kafka 写入 hive - 卡住

【英文标题】：Writing data from kafka to hive using pyspark - stucked

【发布时间】：2020-02-16 16:15:31

【问题描述】：

我刚接触 spark 并开始使用 pyspark，我正在学习使用 pyspark 将数据从 kafka 推送到 hive。

from pyspark.sql import SparkSession
from pyspark.sql.functions import explode
from pyspark.sql.functions import *
from pyspark.streaming.kafka import KafkaUtils
from os.path import abspath

warehouseLocation = abspath("spark-warehouse")

spark = SparkSession.builder.appName("sparkstreaming").getOrCreate()

df = spark.read.format("kafka").option("startingoffsets", "earliest").option("kafka.bootstrap.servers", "kafka-server1:66,kafka-server2:66").option("kafka.security.protocol", "SSL").option("kafka.ssl.keystore.location", "mykeystore.jks").option("kafka.ssl.keystore.password","mykeystorepassword").option("subscribe","json_stream").load().selectExpr("CAST(value AS STRING)")

json_schema = df.schema

df1 = df.select($"value").select(from_json,json_schema).alias("data").select("data.*")

上述方法不起作用，但是在提取数据后，我想将数据插入到 hive 表中。

由于我是全新的，寻求帮助。 提前表扬！ :)

【参考方案1】：

from os.path import expanduser, join, abspath

from pyspark.sql import SparkSession
from pyspark.sql import Row

# warehouse_location points to the default location for managed databases and tables
warehouse_location = abspath('spark-warehouse')

spark = SparkSession \
.builder \
.appName("Python Spark SQL Hive integration example") \
.config("spark.sql.warehouse.dir", warehouse_location) \
.enableHiveSupport() \
.getOrCreate()

# spark is an existing SparkSession
spark.sql("CREATE TABLE IF NOT EXISTS src (key INT, value STRING) USING hive")
spark.sql("LOAD DATA LOCAL INPATH 'examples/src/main/resources/kv1.txt' INTO TABLE src")

【讨论】：

我在 kafka 中的值是一个 json 流，比如 "foo":"foo", "bar":"bar","table_name":"mytable" "barfoo": "barfoo","foobar":"foobar","table_name":"mytable" 一旦我做了 df.selectExpr("CAST (value AS STRING)"), 我如何将数据插入到 hive 表 "mytable",你能帮忙吗

我编辑了代码，如果您仍然遇到一些问题，请检查并实施，请告诉我

@Jay Kakadiya 你的意思是将数据从 kafka 保存到 hdfs 位置，然后将其加载到临时表以插入到真实表中，如果我的理解不正确，请告诉我