我的项目是，写json到Kafka主题，并从kafka主题读取json，最后下沉一个csv。一切都很好。但有些关键字是嵌套的json。如何解析json中的列表？示例Json： {"a": "test", "b": "1234", "c": "temp", "d": [{"test1": "car", "
浏览 14提问于2020-01-12得票数 0
向Kafka topic写入json，并从kafka Topic读取json。实际上，我是逐行订阅主题和编写控制台的。但我必须接收/写入文件csv。但是我不能。我只写了一次csv，但没有附加。import pysparkfrom pyspark.sql.types import * .load()\
df_query = df
浏览 23提问于2020-01-11得票数 1
回答已采纳
1回答
我正在使用Kafka流式传输一个JSON文件，将每一行作为一条消息发送。其中一个关键字是用户的email。from pyspark import SparkContextfrom pyspark.streaming.kafkaimport KafkaUtils
import <em
浏览 2提问于2018-11-21得票数 0
/pyspark/sql/streaming.py"，行827，在start File "/home/b3ds/hdp/spark/python/lib/py4j-0.10.4-src.zip/py4jimport udffrom pyspark import SparkContext
from pyspark<
浏览 4提问于2018-06-04得票数 2
我正在尝试从我的Kafka集群中检索tweet到Spark Streaming，在Spark Streaming中我执行一些分析，将它们存储在ElasticSearch索引中。import osfrom pyspark.stre
浏览 82提问于2020-11-04得票数 2
下面是我的jupyter代码：os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-streaming-kafka-0-8_2.11:2.0.2 pyspark-shell'from pyspark import SparkContext
from pyspark.streamingim
浏览 0提问于2019-12-19得票数 1
我可以使用另一个python脚本从事件中心获取消息，但不能使用Pyspark流式传输消息。下面是我用来流式传输消息的Pyspark代码： connectionString = <MyConnectionString>
'eventhubs.connectionStringoptions(**ehConf).load()
df.writeStream.format("parquet").outputMode("append").option("path&quo
浏览 42提问于2019-12-04得票数 0
1回答
卡夫卡的电火花印刷信息
、、、
我建立了一个由生产者和消费者组成的kafka系统，将json文件的行作为消息流。from pyspark.sql import SQLContext, Row .setAppName("<e
浏览 0提问于2018-11-18得票数 1
回答已采纳
spark-submit --packages org.apache.spark:spark-streaming-kafka-0-8_2.11:2.4.7 Streaming Example.pyfrom pyspark import SparkContextfrom pyspark.streaming importStreamingContext
from pyspark.st
浏览 15提问于2021-05-27得票数 0
我将遵循本教程：Getting Started with Spark Streaming with Python and Kafka 我正在将代码复制到我的Jupiter Notebook (Python完整的代码： #1os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-streaming-kafka-0-8_2.11:2.0.2 pyspark-shell'
f
浏览 36提问于2019-05-16得票数 0
回答已采纳
我正在从kafka流式传输json数据，我需要转换为pyspark中的Dataframe。为了流媒体，我使用了下面的代码。from __future__ import print_functionimport csvfrom pyspark import SparkContextfrom pyspark.streaming import StreamingContext
from pyspark.streamin
浏览 3提问于2017-07-30得票数 0
我正在尝试从一个启用了Kafka兼容性的Azure Eventhub中使用Pyspark解析JSON消息。我找不到任何有关如何建立连接的文档。import osfrom pyspark.streaming import StreamingContextimport json
sc.stop(
浏览 9提问于2019-08-19得票数 0
回答已采纳
我正在尝试编写一个简单的pyspark作业，它将从kafka broker主题接收数据，对这些数据进行一些转换，并将转换后的数据放在不同的kafka broker主题上。我有以下代码，它从kafka主题读取数据，但在运行sendkafka函数时没有任何效果：import sysfrom <e
浏览 4提问于2016-05-20得票数 8
回答已采纳
1回答
我正试图通过Dataproc提交一个pyspark作业，并不断得到一个错误，看起来它没有加载kafka流包。"jobId": "job-33ab811a" }, "pysparkJob": { "mainPythonFileUri": "gs://projectname/streaming.pyorg.apache.spark:spark-sql-kafka-0-10_2.11:2.2.0"
浏览 0提问于2018-03-11得票数 0
回答已采纳
1回答
我想通过火花流从文件夹位置读取json数据。我假设我的json数据是我希望Spark表中的输出为：--import org.apache.spark.streaming.{Spark
浏览 4提问于2017-07-22得票数 2
from kafka import KafkaProduceruserdata={from __future__ import print_functionimport sys
impor
浏览 9提问于2017-08-28得票数 0
回答已采纳
from pyspark.sql import SparkSessionfrom pyspark.streaming importStreamingContextimport json
sc = SparkContext('local/pyspark/
浏览 37提问于2019-12-16得票数 0
下面的代码帮助我做到这一点：import sysfrom pyspark.sql.types import *from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import，它说它需要DataFrame存储在Mo
浏览 2提问于2018-02-22得票数 1
2回答
我是Spark Streaming的新手，我正在尝试使用pyspark读取和解析来自Kafka的JSON流。读取流是可以的，而且我也可以pprint() RDDs。,"WattsLimit":"-1"}但我不知道如何对它们使用json.loads()。我的python代码是：
i
浏览 3提问于2016-03-04得票数 1
云服务器
ICP备案
云直播
腾讯会议
对象存储
腾讯云开发者
