1回答
目标：不断地将嗅探的网络包提供给卡夫卡生产商，将其与星火流连接起来，以便能够处理包数据，然后使用Tensorflow或Keras中的预处理数据。我正在处理来自卡夫卡的火花流(PySpark)中的连续数据，现在我想将处理过的数据发送到Tensorflow。如何用Python在Tensorflow
浏览 0提问于2018-12-18得票数 3
我想处理从Kafka流到PySpark的日志数据并保存到拼图文件中，但我不知道如何将数据输入到Spark。请帮帮我谢谢。
浏览 68提问于2020-07-04得票数 0
1回答
我已经开发了一个databricks管道，它使用Kafka主题中的数据。目前，我正在通过在Pyspark中手动定义模式来进行处理。我知道Kafka模式注册表也是由Kafka团队设置的，我收到了注册表URL。
请指导我如何通过Databricks中的pyspark程序连接到模式注册表。谢谢
浏览 12提问于2021-05-20得票数 1
我有一个Kafka流作业(Spark version2.4.5)在kubernetes上运行，有一个驱动程序和4个executors.The，该作业的目的是消费来自PySpark主题的数据，并对它们进行处理问题：
有时，我们会遇到Kafka集群的问题。因此，PySpark流作业将不会获得任何要处理/消费的数据，即acceptable.However，即使
浏览 15提问于2021-11-25得票数 1
1回答
我正在使用PySpark的DataFrame部分来分析来自Apache Kafka的数据。我遇到一些麻烦，需要一些帮助。from pyspark.sql import functions
# selected_df is dataframe come from kafka use spark.readStream.format("kafka")...").groupBy(fun
浏览 21提问于2019-03-14得票数 1
1回答
我正在尝试用Spark实现并行处理。我想在spark中创建多个接收器(而不仅仅是线程)来接收来自kafka的流数据。我找到了一个链接，说明了如何使用scala实现这一点(参见下面的链接)。但我找不到类似的pyspark代码。有人能帮帮忙吗？
浏览 2提问于2015-09-09得票数 1
2回答
为了捕捉Server中的数据更改，我尝试执行从Server到Pyspark的管道，我已经准备好了一切：
从SQL Server到Kafka进行生产，并使用Pyspark结构化流中的Kafka主题。问题是:当我试图用控制台使用者检查数据更改是否经过Kafka时，它会向我显示JSON格式的消息，分为两个记录: Schema和Paylo
浏览 0提问于2021-06-30得票数 0
回答已采纳
卡夫卡有许多读/写火花数据的方法。我试着阅读来自卡夫卡主题的信息，并从中创建一个数据框架。可以从主题中提取消息，但无法将其转换为数据文件。任何建议都会有帮助。import pysparkfrom pyspark.context import SparkContextdf = spark \
浏览 1提问于2020-06-12得票数 1
我在Python中使用Spark3.0.0。我有一个卡夫卡的test_topic，正在生产从一个csv。import jsonfrom kafka import KafkaConsumerfrom<kafka.consumer.group.KafkaConsumer at 0x13bf55b0>
如何编辑上面的代码给我一个DStream？我是新来的</e
浏览 5提问于2020-08-06得票数 1
回答已采纳
1回答
我第一次使用火花放电。火花版本: 2.3.0卡夫卡版本: 2.2.0
我有一个卡夫卡生产者，它以avro格式发送嵌套数据，我正在尝试用火花流/结构化流写代码，这将反序列化从kafka到dataframe的avro，做转换，把它以拼花格式写成我能够在spark/scala中找到avro转换器，但是还没有增加对pyspark的支持。如何在火花放电中转换相同的内容。谢谢。
浏览 3提问于2019-05-02得票数 3
回答已采纳
我将使用pyspark进行流处理，并使用Kafka作为数据源。
我发现Spark Python API不支持Kafka 0.10连接器。我可以在Spark 2.3.0中使用Kafka 0.8连接器吗？
浏览 14提问于2018-03-02得票数 4
1回答
我正在尝试用spark做一个结构化的流媒体应用程序，主要思想是从kafka源读取，处理输入，写回另一个主题。我已经成功地让spark读写了kafka，但是我的问题出在处理部分。我尝试过foreach函数来捕获每一行并在写回kafka之前对其进行处理，但是它总是只做foreach部分，而不会写回kafka。但是，如果我从writestream中删除foreach部分，它将继续写入，但现在我丢失了处理。
如果有人能给我举
浏览 37提问于2019-07-17得票数 0
回答已采纳
1回答
我对kafka非常陌生，我试图将数据写到一个主题中，并从同一个主题中读取数据(我们现在作为一个源团队来摄取数据)。因此，我们在对Kafk主题进行写操作，并从相同的主题中使用)。&我们还可以指定要将数据拆分为使用的分区数。代码中看到了自定义分区类，但我使用的是火花流，甚至不确定如何集成它。我对Kafka主题分区的困惑有以下几点：
<em
浏览 3提问于2021-09-08得票数 1
回答已采纳
如何更改或完全抑制此批处理元数据而仅显示我的内容？import SparkSessionimport time .builder \
.appName("PySpark Structured Streaming with
浏览 10提问于2020-07-27得票数 1
我想为我的pyspark应用程序使用kafka自定义分割器，从kafka推送到另一个kafka主题。使用pyspark处理将数据从源转换到宿。我希望能够根据data/message中的某个键控制应该将数据推送到哪个分区。在中，我找不到此类用例的任何参考资料或示例。我正在使用python处理和pyspark</e
浏览 1提问于2021-11-01得票数 1
我对spark非常陌生，并从pyspark开始，我正在学习使用pyspark将数据从kafka推送到hive。from pyspark.sql import SparkSessionfrom pyspark.sql.functionsimport *
from pyspark.streaming.kafka import Kafka
浏览 2提问于2020-02-17得票数 0
对于我的当前项目，我正在使用Kafka (python)，并想知道是否有任何方法可以将流式Kafka数据发送到AWS S3桶(而不使用汇流)。我从Reddit API获得我的源数据。我甚至想知道Kafka+s3是否是一个很好的组合，用于存储数据，这些数据将使用pyspark进行处理，或者我应该跳过s3步骤，直接从Kafka读取<
浏览 5提问于2022-08-28得票数 0
2回答
我正在尝试读取通过Kafka和SparkStreaming获取的HDFS存储的数据。我使用的是一个Java应用程序，它使用JavaRDD.saveAsTextFile方法将一些任意数据保存到Hadoop。数据保存在本地主机:50070的默认hadoop浏览器中。
然后，在pyspark应用程序中，我尝试使用sparkContext.textFile读取数据。问题是
浏览 0提问于2018-01-22得票数 0
回答已采纳
应用程序的生产者将产生一些关于特定主题的数据。使用者将使用来自同一主题的数据，并使用spark对其进行处理，并存储这些数据为cassandra表。输入的数据以字符串格式开始，如下所示-
100=NO|101=III|102=0.0771387731911|103=-0.7076915761 100=NO|101=AAA|102=0.8961325446464|10
浏览 2提问于2017-04-04得票数 1
目前我有一些pyspark代码，它是从kafka主题读取数据(readStream)，我计划使用事务更新SQL Server表。流数据将具有所有三个插入、更新、删除事务。
浏览 2提问于2020-01-29得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
云直播
对象存储
实时音视频活动推荐
腾讯云开发者
