Pyspark使用kafka读取现有记录

文章/答案/技术大牛

发布

1回答

apache-spark、pyspark、apache-kafka、spark-streaming

我已经将我的Postgres DB连接到Kafka，以读取表中的新记录，并将它们推送到elasticsearch。数据库已经有一些记录，当我将kafka连接到数据库时，这些记录在kafka主题中可见(使用./kafka-console-consumer.sh --topic postgres.public.table --bootstrap-server kafka:9092 --from-beginnin

浏览 71提问于2021-10-07得票数 0

1回答

在pyspark中将Kafka的结果写入csv

apache-spark、pyspark、apache-kafka

我已经设置了一个Kafka代理，并且我设法用pyspark读取记录。import osimport pysparkfrom pyspark import SparkConf, SparkContext, SQLContext, Rowfrom pysp

浏览 12提问于2020-05-20得票数 0

1回答

Spark流式数据以更新SQL Server (事务)

sql-server、spark-streaming

目前我有一些pyspark代码，它是从kafka主题读取数据(readStream)，我计划使用事务更新SQL Server表。流数据将具有所有三个插入、更新、删除事务。

浏览 2提问于2020-01-29得票数 0

2回答

使用pyspark批处理作业读取Kafka的最新记录

apache-spark、apache-kafka

我正在pyspark中执行一个批处理作业，其中spark将每5分钟读取一次kafka主题中的数据。df = spark \ .format("kafka") \ ..option("startingOffsets", "earliest") \ .option(&quo

浏览 2提问于2020-03-04得票数 0

2回答

pyspark是否支持spark-streaming-kafka-0-10 lib？

pyspark、apache-kafka、spark-streaming-kafka

我的kafka集群版本是0.10.0.0，我想使用pyspark流来读取kafka数据。但是在Spark Streaming + Kafka集成指南中，没有python代码示例。那么pyspark可以使用spark-streaming kafka -0-10来集成kafka吗？提前感谢您的帮助！

浏览 11提问于2017-08-05得票数 0

2回答

无法使用spark streaming连接到kafka主题(python、jupyter)

python、apache-spark、pyspark、apache-kafka

我试着用spark连接到kafka的话题。它不会读取数据流中的任何数据，也不会产生任何错误。下面是我的jupyter代码：os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-streaming-kafka-0-8_2.11:2.0.2 pyspark-shell'from pyspark</em

浏览 0提问于2019-12-19得票数 1

1回答

TypeError：'JavaPackage‘对象不可调用&在类路径中找不到Spark Streaming的Kafka库

apache-spark、pyspark、apache-kafka、spark-streaming

我使用pyspark流来读取kafka数据，但它出错了：from pyspark.streaming.kafka import KafkaUtilssc = SparkContext(appName="test"

浏览 0提问于2020-01-05得票数 0

1回答

无法使用spark streaming从kafka主题读取数据

apache-spark、apache-kafka、spark-streaming-kafka、kafka-topic

我正在尝试使用spark streaming来读取kafka主题中的数据。我可以将消息生成kafka主题，但当使用spark streaming读取主题中的数据时，我收到错误消息，如下所示：代码如下： from

浏览 0提问于2019-05-27得票数 1

1回答

如何在码头形象中安装卡夫卡？

docker、apache-spark、apache-kafka、docker-compose

我的Dockerfile内容： - SPARK_LOCAL_STORAGE_ENCRYPTION_ENABLED=no我正在尝试使用tweepy库读取tweet，并将其发送给Kafka &<em

浏览 8提问于2022-04-03得票数 0

1回答

使用DStream接口从Kafka读取Spark Straming中的时间戳

python、apache-spark、pyspark、apache-kafka、spark-streaming

我想从Kafka主题中读取使用Python的Spark streaming的值。我使用的是带有spark-streaming-kafka-0-8支持的DStream应用程序接口(尽管已被弃用)。我的代码如下： from pyspark import SparkContext, SparkConffrompyspark.streaming.kafka<

浏览 22提问于2020-04-30得票数 0

1回答

pyspark kafka流偏移量

apache-spark、pyspark、apache-kafka、streaming、offset

我从下面的链接中获得了关于kafka主题偏移流的pyspark。from pyspark.streaming import StreamingContextfrom pyspark.streaming.kafkatopicPartion = TopicAndPartition(topic, 0) fromOffset = {topicPartion: long(PUT NUM

浏览 0提问于2018-10-06得票数 0

2回答

如何使用pyspark读取hdfs kafka数据？

java、python、hadoop、apache-spark、pyspark

我正在尝试读取通过Kafka和SparkStreaming获取的HDFS存储的数据。我使用的是一个Java应用程序，它使用JavaRDD.saveAsTextFile方法将一些任意数据保存到Hadoop。然后，在pyspark应用程序中，我尝试使用sparkContext.textFile读取数据。问题是我读取的数据(使用python或localhost:50070中的“手工”)也包含元数据。偏移量= 3，CreateTime =

浏览 0提问于2018-01-22得票数 0

回答已采纳

3回答

如何在星火流应用程序中使用Kafka主题？

apache-spark、pyspark、apache-kafka、spark-streaming

当我从Kafka主题创建一个流并打印它的内容时 os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.apache.spark:spark-streaming-kafka-0-8_2.11:2.0.2 pyspark-shell' from pyspark.streamingimpo

浏览 2提问于2019-12-08得票数 1

回答已采纳

1回答

如何正确使用pyspark向kafka* broker发送数据？*

python-2.7、pyspark、spark-streaming、kafka-python

我正在尝试编写一个简单的pyspark作业，它将从kafka broker主题接收数据，对这些数据进行一些转换，并将转换后的数据放在不同的kafka broker主题上。我有以下代码，它从kafka主题读取数据，但在运行sendkafka函数时没有任何效果：import sys from pyspark.streaming impo

浏览 4提问于2016-05-20得票数 8

回答已采纳

2回答

如何使用pyspark将spark转换的数据写回kafka代理？

python-2.7、pyspark、spark-streaming、kafka-producer-api、kafka-python

在我的pyspark应用程序中，我打算使用Spark streaming作为一种“飞行中”转换Kafka消息的方法。每个这样的消息最初都是从特定的Kafka主题接收的。这样的消息需要经过一些转换(比方说-用一个字符串替换另一个字符串)，转换后的版本需要发布在不同的Kafka主题上。第一部分(接收Kafka消息)似乎工作正常：imp

浏览 1提问于2016-05-20得票数 1

2回答

Spark流处理二进制数据文件

pyspark、spark-streaming

我使用的是pyspark 1.6.0。在PYSPARK中，我使用: sc.binaryFiles("s3n://.......")读取二进制文件。这很有效，因为它提供了一个元组(文件名和数据)，但我正在尝试找到一个等效的PYSPARK流API来读取二进制

浏览 3提问于2016-06-29得票数 6

1回答

pySpark:将Kafka流放入parquet中，并从远程会话读取parquet

python、docker、pyspark、apache-kafka、parquet

Spark读取Kafka流。stream_df = spark.readStream.format("kafka")\ .option("kafka.bootstrap.servers", "kafka1:19091")\从不同的docker容器中，我尝试读取拼图文件。File "/usr/local/lib/python3.9/site-packages&#x

浏览 36提问于2021-04-29得票数 0

1回答

使用kafka作为外接程序的Heroku postgresql的火花流

python、postgresql、heroku、pyspark、apache-kafka-streams

我正在尝试使用数据库中的pyspark从Heroku postgres数据库流，使用kafka作为插件。下面有这样的代码，但是它连续运行，没有返回任何内容，也没有错误。我尝试在代码中添加触发器，但不知道要使用什么导入，因为我没有在网上找到它的包。.{“消息”：“初始化源”，“isDataAvailable”：False，“isTriggerActive”：False} 我使用了kafka-python，并且能够读取流，但是使用<em

浏览 14提问于2022-10-16得票数 0

1回答

在火花流/结构化流中阅读Kafka的avro消息

pyspark、apache-kafka、spark-streaming、spark-structured-streaming、spark-streaming-kafka

我第一次使用火花放电。火花版本: 2.3.0卡夫卡版本: 2.2.0 我有一个卡夫卡生产者，它以avro格式发送嵌套数据，我正在尝试用火花流/结构化流写代码，这将反序列化从kafka到dataframe的avro，做转换，把它以拼花格式写成我能够在spark/scala中找到avro转换器，但是还没有增加对pyspark的支持。如何在火花放电中转换相同的内容。谢谢。

浏览 3提问于2019-05-02得票数 3

回答已采纳

2回答

只捕捉到的有效载荷的CDC在火花结构化的流？

sql-server、apache-spark、pyspark、apache-kafka、cdc

为了捕捉Server中的数据更改，我尝试执行从Server到Pyspark的管道，我已经准备好了一切： import os os.en

浏览 0提问于2021-06-30得票数 0

回答已采纳

点击加载更多