我已经设置了一个Kafka代理，并且我设法用pyspark读取记录。, SparkContext, SQLContext, Rowfrom pyspark.streaming.kafkassc.start()返回数据示例(时间戳，名称，姓氏，高度)：我想将这些记录写入</em
浏览 12提问于2020-05-20得票数 0
向Kafka topic写入json，并从kafka Topic读取json。实际上，我是逐行订阅主题和编写控制台的。但我必须接收/写入文件csv。但是我不能。我只写了一次csv，但没有附加。你可以在下面看到我的代码。 谢谢!import pysparkfrom pyspark.sql.types import *
浏览 23提问于2020-01-11得票数 1
回答已采纳
1回答
我正在尝试用spark做一个结构化的流媒体应用程序，主要思想是从kafka源读取，处理输入，写回另一个主题。我已经成功地让spark读写了kafka，但是我的问题出在处理部分。我尝试过foreach函数来捕获每一行并在写回kafka之前对其进行处理，但是它总是只做foreach部分，而不会写回kafka。但是，如果我从writestream中删除foreach部分，它将继续写入，但现在我丢失了处理。
如果有人能给我举一个例子，告诉我如何做到这一点，我将非常感激。以下是我
浏览 37提问于2019-07-17得票数 0
回答已采纳
目前我有一些pyspark代码，它是从kafka主题读取数据(readStream)，我计划使用事务更新SQL Server表。流数据将具有所有三个插入、更新、删除事务。
浏览 2提问于2020-01-29得票数 0
1回答
有没有人尝试使用spark-steaming(pyspark)作为CDH中kerberos KAFKA的消费者？是不是CDH不支持？
浏览 2提问于2018-02-08得票数 1
简单地说，有没有可能将DStream流式传输到Kafka主题？
我有火花流作业，它做了所有的数据处理，现在我想把数据推送到卡夫卡主题。在pyspark中可以做到这一点吗？
浏览 1提问于2018-06-08得票数 0
2回答
我正在尝试读取通过Kafka和SparkStreaming获取的HDFS存储的数据。数据保存在本地主机:50070的默认hadoop浏览器中。
然后，在pyspark应用程序中，我尝试使用sparkContext.textFile读取数据。问题是我读取的数据(使用python或localhost:50070中的“手工”)也包含元数据。然后拆分和解析长字符串以获得“值”内容并不是最好的主意。还是拯救本身的问题？
浏览 0提问于2018-01-22得票数 0
回答已采纳
2回答
火花放电版本- 2.4.7卡夫卡版本- 2.13_3.2.0findspark.init("/usr/local/spark")
from pyspark.sql import S
浏览 21提问于2022-06-13得票数 0
Spark读取Kafka流。stream_df = spark.readStream.format("kafka")\
.option("kafka.bootstrap.servers", "kafka1:19091")\并将stream_df写入拼图文件。从不同的docker容器中，我尝试读取拼图文件。但是如果提交一个读取spark容器上拼花文件的应用程序，我就可以访问数据... 我已经尝试了
浏览 36提问于2021-04-29得票数 0
我有一个代码(kafka_producer.py)，从csv >>中读取，创建Pandas dataframe >>，将熊猫的数据转换为星星之火>>调用>>方法，将消息发送给kafka。df = pd.read_csv(path,sep=sep) print("Converting Pandas(df)
浏览 37提问于2022-05-27得票数 0
回答已采纳
我使用pyspark流从kafka获取数据，结果是一个dataframe，当我将dataframe转换为rdd时，它出错了： Traceback (most recent call last):pyspark.sql.utils.AnalysisException: 'Queries with streamingsources must be executed wi
浏览 56提问于2020-01-06得票数 3
1回答
火花版本: 2.3.0卡夫卡版本: 2.2.0
我有一个卡夫卡生产者，它以avro格式发送嵌套数据，我正在尝试用火花流/结构化流写代码，这将反序列化从kafka到dataframe的avro，做转换，把它以拼花格式写成我能够在spark/scala中找到avro转换器，但是还没有增加对pyspark的支持。如何在火花放电中转换相同的内容。谢谢。
浏览 3提问于2019-05-02得票数 3
回答已采纳
2回答
我可以在终端机运行这个。我的终端命令是：import os
os.environ['PYSPARK_SUBMIT_ARGS'] = "--packages org
浏览 3提问于2018-04-16得票数 1
我的项目是，写json到Kafka主题，并从kafka主题读取json，最后下沉一个csv。一切都很好。但有些关键字是嵌套的json。如何解析json中的列表？temp", "d": [{"test1": "car", "test2": 345}, {"test3": "animal", "test4": 1}], "e"
浏览 14提问于2020-01-12得票数 0
1回答
我有大约500-700 csv文件的日志加载每周。文件名是每周随机生成的。这些文件存储在google云存储上，pyspark在google dataproc上运行，有几个工作节点。我以为我会在bigquery表上看到增量加载，但似乎pyspark会将整个gz输入文件合并到dataframe中，然后立即处理它。所以我在大约90分钟后得到了结果。我如何编写pyspark代码，以便每次读取文件时，都会将其写入</e
浏览 22提问于2021-10-10得票数 1
我计划使用Spark结构化流从Kafka中提取数据，但我得到了空数据。# -*- coding: utf-8 -*-from pyspark.sql.functions import from_csv= SparkSession \ .appName("pyspark_structured_streaming_kafka") \Laptop
浏览 2提问于2021-02-04得票数 1
回答已采纳
只需设置一个hadoop/kafka/星火，一个节点演示环境。在pyspark中，我尝试读取(.readStream) Kafka消息，并在hadoop中将它写入(.writeStream)到json文件。奇怪的是，在hadoop "output/test“目录下，我可以找到一个创建的json文件，但只有一个消息。卡夫卡的所有新消息都不会更新json文件。但是我想把卡夫卡<em
浏览 2提问于2022-08-27得票数 0
我编写了一些连接到kafka broker的示例代码，从主题读取数据并将其放入snappydata表。from pyspark.conf import SparkConffrom pyspark.sql import SQLContext主题中读取，并在snappydata表中写入。我不明白为什么我没有在SnappyData仪表板UI中看到这个流查询--在控制台中提交了
浏览 1提问于2020-11-30得票数 0
回答已采纳
我是卡夫卡和斯帕克的新手。我已经通过Kafka生产者传递了消息，并试图在spark流中读取，但在main方法中遇到错误。代码如下所示。spark-submit --packages org.apache.spark:spark-streaming-kafka-0-8_2.11:2.4.7 Streaming Example.pyfrom pyspark import SparkContextfrom <e
浏览 15提问于2021-05-27得票数 0
我正在尝试从一个启用了Kafka兼容性的Azure Eventhub中使用Pyspark解析JSON消息。我找不到任何有关如何建立连接的文档。import osfrom pyspark.streaming import StreamingContext
from pyspark.streaming.kafka
浏览 9提问于2019-08-19得票数 0
回答已采纳
云服务器
ICP备案
实时音视频
对象存储
云直播
腾讯云开发者
