PySpark流+卡夫卡字数统计不打印任何结果
PySpark是一种基于Python的Spark编程框架，用于处理大规模数据集的分布式计算。它提供了丰富的API和工具，可以进行数据处理、机器学习、图计算等任务。
流式处理是一种实时处理数据的方式，它能够接收连续的数据流并进行实时处理。PySpark流式处理模块可以通过集成Apache Kafka来实现对数据流的处理。
Apache Kafka是一种高吞吐量、可扩展的分布式流处理平台，用于构建实时数据流应用程序和数据管道。它具有持久性、可靠性和容错性，并能够处理大规模的数据流。
卡夫卡字数统计是指使用PySpark流+卡夫卡来实现对数据流中文本内容的字数统计。具体步骤如下：
- 配置和启动Kafka集群：使用Kafka提供的命令行工具或API，配置和启动一个Kafka集群，包括创建主题(topic)用于接收数据流。
- 编写PySpark流式处理代码：使用PySpark编写流式处理代码，包括连接到Kafka集群、读取数据流、进行字数统计等操作。
- 发送数据到Kafka主题：将需要进行字数统计的文本数据发送到Kafka主题中，可以使用Kafka的生产者API或其他工具。
- 实时字数统计：PySpark流式处理代码会实时接收Kafka主题中的数据流，并进行字数统计。可以使用PySpark提供的函数和操作符来实现字数统计功能。
- 结果输出：根据需求，可以选择将字数统计结果保存到数据库、写入文件或发送到其他系统进行进一步处理。
推荐的腾讯云相关产品和产品介绍链接地址如下：
- 腾讯云消息队列 CKafka：https://cloud.tencent.com/product/ckafka 腾讯云提供的高可用、高可靠的消息队列服务，可用于构建实时数据流应用程序。
- 腾讯云云服务器 CVM：https://cloud.tencent.com/product/cvm 腾讯云提供的弹性计算服务，可用于部署和运行PySpark流式处理代码。
请注意，以上仅为示例推荐的腾讯云产品，并非广告推广。在实际应用中，您可以根据具体需求选择适合的产品和服务。
相关·内容
1回答
与卡夫卡的直接流？
、、、、
Pyspark 2.4.7包含创建卡夫卡主题()的直接流侦听器的能力。
但是，3.1.1 (最新)版本的pyspark没有这个特性。卡夫卡直接流的不再包括python示例。
我的问题是如何使用直接流(而不是结构化流)与kafka和pyspark 3.1.1？
浏览 2提问于2021-05-06得票数 1
回答已采纳
3回答
当我从Kafka主题创建一个流并打印它的内容时
import os
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.streaming import StreamingContext
from pyspark.streaming.kafka impo
浏览 2提问于2019-12-08得票数 1
回答已采纳
2回答
如何将火花流与卡桑德拉连接起来？
、、、、
我在用
Cassandra v2.1.12
Spark v1.4.1
Scala 2.10
卡桑德拉正在收听
rpc_address:127.0.1.1
rpc_port:9160
例如，要连接卡夫卡和火花流，同时每4秒听一次卡夫卡，我有下面的火花作业
sc = SparkContext(conf=conf)
stream=StreamingContext(sc,4)
map1={'topic_name':1}
kafkaStream = KafkaUtils.createStream(stream, 'localhost:2181', "name
浏览 5提问于2016-01-25得票数 3
卡夫卡有许多读/写火花数据的方法。我试着阅读来自卡夫卡主题的信息，并从中创建一个数据框架。可以从主题中提取消息，但无法将其转换为数据文件。任何建议都会有帮助。
import pyspark
from pyspark.sql import SparkSession, Row
from pyspark.context import SparkContext
from kafka import KafkaConsumer
sc = SparkContext.getOrCreate()
spark = SparkSession(sc)
consumer = KafkaConsumer('J
浏览 1提问于2020-06-12得票数 1
1回答
火花流+ Kafka集成
、、、、
我试图整合星火和卡夫卡在木星笔记本使用。这是我的工作环境。
火花版本:星星之火2.2.1卡夫卡版本: Kafka_2.11-0.8.2.2火花流卡夫卡jar: spark-streaming-kafka-0-8-assembly_2.11-2.2.1.jar
我添加了一个streaming程序集jar文件来激发-defaults.conf文件。
当我为火花放电流启动streamingContext时，此错误将显示为无法从MANIFEST.MF.中读取卡夫卡版本。
这是我的密码。
from pyspark import SparkContext, SparkConf
from pysp
浏览 1提问于2018-08-08得票数 0
2回答
我是作为JSON从kafka_2.11-0.10.0.1到Spark2.1的流式计量器读取记录。我转向了结构化流媒体；尽管卡夫卡用户确认了输入的数据，但我的控制台和writeStream没有移动。我正在测试使用
pyspark --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.1.0
我的代码：
from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *
spark = SparkSe
浏览 0提问于2017-12-15得票数 3
回答已采纳
我有一个pyspark应用程序，它配置了2个数据流：
数据流A)从卡夫卡(主题1和2)读取的(主题1和2) ->合并主题DStreams ->做的东西->输出到REDIS。
数据流B)从卡夫卡(主题3、4和5)读取的(主题3、4和5) ->将主题DStreams ->做的内容->输出合并到相同的REDIS。
只有当其中一个配置好，而不是两者都配置时，我才能在REDIS中获取数据。
实际上，即使我将REDIS输出更改为pprint，也会发生这种情况。只有当两个流中只有一个处于活动状态时，我才会打印输出。
我错过了什么？
浏览 0提问于2016-02-10得票数 2
2回答
卡夫卡消费者的火花流
、、、、
我可能需要和卡夫卡合作，而且我对此完全陌生。据我所知，卡夫卡的制片人会将日志(称为事件或信息或卡夫卡中的记录)发布到卡夫卡的主题上。
我需要通过消费者阅读卡夫卡的主题。我需要先设置消费API，然后才能使用SparkStreaming上下文(PySpark)进行流，还是可以直接使用KafkaUtils模块从KafkaUtils主题中读取？
如果我需要设置Kafka消费者应用程序，我如何做到这一点？请您共享指向正确文档的链接。
提前谢谢！！
浏览 1提问于2016-07-01得票数 5
回答已采纳
我是卡夫卡和斯帕克的新手。我已经通过Kafka生产者传递了消息，并试图在spark流中读取，但在main方法中遇到错误。代码如下所示。
spark-submit --packages org.apache.spark:spark-streaming-kafka-0-8_2.11:2.4.7 Streaming Example.py
from pyspark import SparkContext
from pyspark.sql import SparkSession
from pyspark.streaming import StreamingContext
from pyspar
浏览 15提问于2021-05-27得票数 0
我是个新手，正在使用PySpark。我注意到字数统计程序的PySpark的DAG与Scala Spark的不同。有人能帮我理解一下吗？
第一个DAG是PySpark
其次是Scala Spark
浏览 3提问于2018-10-16得票数 1
1回答
带有火花流的Kafka抛出了一个错误：
from pyspark.streaming.kafka import KafkaUtils ImportError: No module named kafka
我已经建立了一个卡夫卡经纪人和一个工作火花环境与一个主人和一个工人。
import os
os.environ['PYSPARK_PYTHON'] = '/usr/bin/python2.7'
import findspark
findspark.init('/usr/spark/spark-3.0.0-preview2-bin-hadoop2.7&#
浏览 2提问于2020-05-19得票数 4
回答已采纳
我在试着用卡夫卡来证明火花。我有卡夫卡的消费者有json数据。我想把卡夫卡的消费者和斯派克一起进行加工。当我运行下面的代码时，会抛出错误。
bin\spark-submit --packages org.apache.spark:spark-streaming-kafka-0-8_2.11:2.0.0 test.py localhost:9092 maktest
我的test.py在下面
import sys
from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.
浏览 2提问于2018-12-10得票数 2
回答已采纳
1回答
我有来自卡夫卡主题的数据。在我的dataframe中，我有“平均”列，我希望通过对“平均”列执行一些计算来创建新列。
目前我写的是-
rdd_get_chills = df_avg_tmp.rdd.map(lambda line:get_wind_chills(line))
但这是个错误-
pyspark.sql.utils.AnalysisException: Queries with streaming sources must be executed with writeStream.start();
浏览 3提问于2022-09-20得票数 0
回答已采纳
2回答
当从齐柏林飞艇运行时，我很难让console接收器与一起工作。基本上，我没有看到任何结果打印到屏幕上，也没有打印到我找到的任何日志文件。
，我的问题是：是否有一个使用PySpark结构化流的工作示例，其中的接收器产生了在Apache中可见的输出？理想情况下，它还可以使用套接字源，因为这很容易测试。
我在用：
Ubuntu 16.04
spark-2.2.0-bin-hadoop2.7
zeppelin-0.7.3-bin-all
Python3
我的代码是基于的。它在从PySpark shell (./bin/pyspark --master local[2])运行时工
浏览 3提问于2017-11-17得票数 15
回答已采纳
我正在学习apache spark与kafka的集成，这样当新消息到达Kafka的主题时，我的代码就可以自动运行。 我也读过官方文档 https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html 但我仍然对它的工作原理感到困惑。我有我的字数统计写的程序在java火花和另一边卡夫卡正在运行。 结构化流媒体是kafka和spark java代码之间的桥梁吗？它是否一直在监听kafka，并且每当消息到达时，它都会从kafka中提取消息，并将其传递给java代码？这是对的吗？ 如果没有，有没有人可以和
浏览 15提问于2019-12-10得票数 1
1回答
我想阅读一个相当大的文档，并从中提取N-gram(N个单词的连续序列)。我还想按频率的升序对其进行排序。虽然我知道这是字数统计的一个扩展，但我不确定如何在Pyspark中做到这一点。任何帮助都是非常感谢的。
浏览 0提问于2016-04-14得票数 1
我仍然在开发我用描述的Kafka应用程序。在那篇帖子里，我问为什么
kstreams_props.put( ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
似乎没有把卡夫卡的状态重置为“宇宙的开始”，在任何数据被推到任何话题之前。我现在遇到了这个问题的一个变体：
我的应用程序包括一个生产者程序，它将数据推送到Kafka流，以及一个消费者程序，它将数据分组，聚合组，然后将生成的KTable转换回流，然后我打印出来。
聚合步骤实质上是将所有值加起来，然后将这些和作为新数据放入输出流中。然而，我观察到，每次我运行程序时，产生
浏览 2提问于2018-04-03得票数 0
我想使用结构化流和PySpark一起运行Spark应用程序。
我使用星火2.2和卡夫卡0.10版本。
如果出现以下错误，我将失败：
java.lang.IncompatibleClassChangeError:实现类
spark-submit命令，如下所示：
/bin/spark-submit \
--packages org.apache.spark:spark-streaming-kafka-0-10_2.11:2.2.0 \
--master local[*] \
/home/umar/structured_streaming.py localhost:2181 fo
浏览 5提问于2017-08-22得票数 3
我使用下面的代码将火花流数据流写入到MQSQL中，.Below是卡夫卡主题，JSON数据格式和MYSQL表的schema.Column名称和类型相同。但是我无法看到用MYSQL表编写的记录。表为空，records.Please建议为零。
卡夫卡主题数据
ssingh@RENLTP2N073:/mnt/d/confluent-6.0.0/bin$ ./kafka-控制台-消费者-主题“中等名”：“玫瑰”、“姓氏”：“”、“dob_year”：2010年、"dob_month":3、“性别”：“M”、“工资”：4000}
import pyspark
from pyspark.s
浏览 0提问于2020-11-24得票数 1
回答已采纳
我有一个RDD，它填充为
id txt
1 A B C
2 A B C
1 A B C
我的字数统计(pyspark)的结果应该是字符串和与之关联的id的组合。示例：
[(u'1_A',2), (u'1_B',2), (u'1_C',2),(u'2_A',1),(u'2_B',1),(u'2_C',1)]
我尝试使用用户定义的函数将id与从文本中拆分的字符串组合在一起。然而，它抱怨追加功能在此
浏览 1提问于2017-06-03得票数 0
我在尝试运行字数统计示例时遇到以下错误。
不确定如何继续。下面是我正在运行的命令和错误。
/opt/spark/bin/spark-submit --jars spark-streaming_2.10-2.0.0.jar test_kafka.py broker.txt "localhost:2181:MyTopic"
Error:
Traceback (most recent call last):
File "/home/ubuntu/kafka/libs/test_kafka.py", line 21, in <module
浏览 9提问于2016-08-29得票数 3
2回答
火花流不从本地目录读取
、、、
我正在尝试使用spark编写一个星星之流应用程序。
应用程序应该从本地目录读取文本文件，并将其发送到Kafka集群。
当提交python脚本来激发引擎时，什么也没有发送给kafka。
我试图打印这些事件，而不是把它寄给卡夫卡，结果发现没有什么可读的。
这是脚本的代码。
#!/usr/lib/python
# -*- coding: utf-8 -*-
from pyspark import SparkContext, SparkConf
from pyspark.streaming import StreamingContext
from kafka import KafkaProducer
浏览 2提问于2016-04-17得票数 1
1回答
Kafka流的最佳实践
、、、
我们有一个用python编写的预测服务来提供机器学习服务，您可以发送它一组数据，它会给出异常检测或预测等等。
我想使用Kafka流来处理实时数据。
有两种选择方法：
卡夫卡流作业只完成ETL函数:加载数据，并进行简单的转换和保存数据到弹性搜索。然后启动定时器，周期性地从ES加载数据，并调用预测服务来计算结果并将结果保存回ES。
卡夫卡流作业除了ETL之外，所有的事情都是由卡夫卡流作业完成ETL，然后发送数据预测服务，并将计算结果保存给卡夫卡，消费者将结果从卡夫卡转发到ES。
我认为第二种方法是更实时的，但我不知道在流作业中做这么多的预测任务是个好主意。
对于这种应用，是否有任何
浏览 1提问于2018-08-20得票数 4
1回答
我第一次使用火花放电。火花版本: 2.3.0卡夫卡版本: 2.2.0
我有一个卡夫卡生产者，它以avro格式发送嵌套数据，我正在尝试用火花流/结构化流写代码，这将反序列化从kafka到dataframe的avro，做转换，把它以拼花格式写成s3。我能够在spark/scala中找到avro转换器，但是还没有增加对pyspark的支持。如何在火花放电中转换相同的内容。谢谢。
浏览 3提问于2019-05-02得票数 3
回答已采纳
我想使用星火流，并连接它与卡夫卡。然而，我仍然得到了NoSuchMethodError: NoSuchMethodError错误，现在我只是不知道下一步该做什么。
我的设置：
Ubuntu 16.04
Scala 2.11
Kafka 2.11-1.0.0 (我也尝试使用2.11-0.10.0.0)
火花2.2.1
Hadoop 2.9.0
我甚至不能运行示例脚本：
from pyspark.sql import SparkSession, Row, SQLContext
from pyspark import SparkContext, SparkConf
from pyspark.stre
浏览 1提问于2018-02-09得票数 3
1回答
在从apache网站下载之后，我已经集成了kafka和spark流。然而，我想使用Datastax作为我的大数据解决方案，我看到您可以轻松地集成Cassandra和Spark。
但是，在Datastax企业的最新版本中，我看不到任何kafka模块。如何将卡夫卡与火花流结合在一起？
我想做的基本上是：
启动必要的代理和服务器
启动卡夫卡制片
启动卡夫卡消费者
将火花流连接到kafka broker并接收来自那里的消息。
然而，在google快速搜索之后，我看不到kafka与datastax企业合并的任何地方。
我怎样才能做到这一点？我对datastax和kafka非常陌生
浏览 5提问于2016-01-22得票数 2
回答已采纳
我有一个Kafka流作业(Spark version2.4.5)在kubernetes上运行，有一个驱动程序和4个executors.The，该作业的目的是消费来自PySpark主题的数据，并对它们进行处理，然后将最终数据加载到ElasticSearch。
问题：
有时，我们会遇到Kafka集群的问题。因此，PySpark流作业将不会获得任何要处理/消费的数据，即acceptable.However，即使在Kafka群集变得正常之后，PySpark流作业也不会消耗/处理来自Kafka topic.The作业的任何进一步记录
错误:在驱动程序或任何执行器上找不到错误。
其中一个executors
浏览 15提问于2021-11-25得票数 1
1回答
运行星火流作业时，我已经多次遇到以下行为。处理开始良好:每个批次的处理时间远远低于批处理间隔。然后，突然，输入率下降到接近零。
即使程序能够跟上，并且大大减慢了执行速度，也会发生这种情况。我相信，当没有多少未处理的数据时，就会出现下降，但由于比率太低，这些最终记录占用了运行这项工作所需的大部分时间。有没有办法避免这种情况，加快速度？
我正在使用PySpark与Spark 1.6.2，并使用直接方法的卡夫卡流。Backpressure被打开，maxRatePerPartition为100。
浏览 3提问于2016-08-22得票数 2
1回答
将电火花与卡夫卡连接起来
、、、、
我很难理解卡夫卡和PySpark之间的联系。
我在Windows 10上安装了kafka，其主题是很好的流数据。我已经安装了运行正常的火花放电-我能够创建没有问题的测试DataFrame。
但当我试图连接卡夫卡流时，它会给我带来错误：
AnalysisException:未能找到数据源: kafka。请按照“结构化流-卡夫卡集成指南”的部署部分部署应用程序。
火花文档并不是很有帮助-它说：.groupId = org.apache.spark artifactId =Spark-SQLKafka-0-102.12版本= 3.2.0 .
对于Python应用程序，在部署应用程序时，需要添加上
浏览 6提问于2021-12-16得票数 4
回答已采纳
1回答
目标：不断地将嗅探的网络包提供给卡夫卡生产商，将其与星火流连接起来，以便能够处理包数据，然后使用Tensorflow或Keras中的预处理数据。
我正在处理来自卡夫卡的火花流(PySpark)中的连续数据，现在我想将处理过的数据发送到Tensorflow。如何用Python在Tensorflow中使用这些转换后的DStreams？谢谢。
目前没有任何处理应用于星火流，但将在以后添加。以下是py代码：
import sys
from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from py
浏览 0提问于2018-12-18得票数 3
1回答
我试着用这个代码用broker版本0.10测试kafka流。它只是一个简单的代码来打印一个主题的内容。还没什么大不了的！但是，由于某种原因，内存是不够的(VM中的10 VM内存)！守则：
# coding: utf-8
"""
kafka-test-003.py: test with broker 0.10(new Spark Stream API)
How to run this script?
spark-submit --jars jars/spark-sql-kafka-0-10_2.11-2.3.0.jar,jars/kafka-clients-0.11
浏览 0提问于2018-12-12得票数 1
2回答
如何将记录从卡夫卡传递给方法？
、、、、
我在阅读数据的地方有一个Kafka队列如下所示：
private static void startKafkaConsumerStream() {
try {
System.out.println("Print method: startKafkaConsumerStream");
Dataset<String> lines = (Dataset<String>) _spark
.readStream()
浏览 0提问于2017-09-04得票数 0
回答已采纳
我试图读取数据以将其转换为Dataframe，我的软件的当前版本如下：
spark-2.4.7-bin-hadoop2.7
卡夫卡_2.12-2.7.0
卡夫卡正在工作，我正在努力阅读以下数据：
~/development/kafka_home/kafka_2.13-2.6.0$ bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic testtopic --from-beginning
{"transaction_id": "1", "transac
浏览 0提问于2021-01-25得票数 1
我正在尝试运行word计数示例，它集成了AWS Kinesis流和Apache。随机线按一定的间隔放置在运动中。
lines = KinesisUtils.createStream(...)
当我提交申请时，lines.pprint()没有看到任何值被打印出来。
试图打印lines对象，我看到了<pyspark.streaming.dstream.TransformedDStream object at 0x7fa235724950>
如何打印PythonTransformedDStream对象？检查数据是否收到。
我确信没有凭据问题，如果我使用假凭据，我将获得访问异常。
添加代码
浏览 4提问于2017-01-25得票数 2
回答已采纳
我有以下简单的spark streaming进程，它使用组id为feature1的卡夫卡主题test中的消息并打印结果。但是，当我运行bin/kafka-consumer-groups.sh --bootstrap-server zookeeper-1:9092 --list列出所有组时，没有feature1或任何包含feature1的内容。有什么问题吗？
我的spark版本是2.1.2，kafka版本是2.12-2.0.0，zookeeper版本是3.4.13。我在这里发现了一些相关的问题，但我不知道我的问题是否与这个问题有关。
# coding=utf8
import sys
impor
浏览 9提问于2018-08-20得票数 0
我正在运行一份电火花作业，数据流来自卡夫卡。我试图在我的windows系统中复制一个场景，以了解当消费者下降时，当数据不断被输入Kafka时会发生什么。
这就是我所期望的。
生产者启动并生成消息1、2和3。
使用者是在线的，并且使用消息1、2和3。
现在消费者由于某种原因而下降，而生产者产生消息4、5和6等等……
当消费者出现时，我的期望是它应该读到它停下来的地方。因此，消费者必须能够从第4，5，6等信息中读取.
我的电火花应用程序无法实现我预期的目标。下面是我如何创建星火会话的方法。
session.readStream.format("kafka")
浏览 2提问于2020-05-08得票数 1
回答已采纳
在中，我尝试在笔记本中创建一个kafka流，并使用它来创建一个火花作业。Databricks在KafkaUtils.createDirectStream()行抛出错误。在下面附加了相应的代码。
from kazoo.client import KazooClient
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import KafkaUtils, TopicAndPartition
sc = spark.sparkContext
ssc = StreamingContext(sc, 3
浏览 8提问于2019-11-21得票数 2
回答已采纳
2回答
阿帕奇卡夫卡和火花流
、、、、
我正在阅读这篇博文：
讨论了如何使用Spark流和Apache进行一些近乎实时的处理。我完全理解这篇文章。它确实展示了如何使用星火流来读取来自某个主题的消息。我想知道是否有一个Streaming可以用来将消息写入Kakfa主题？
我的用例非常简单。我有一组数据，我可以在一个固定的间隔(例如每秒钟)从给定的源读取数据。我用反应流来做这件事。我想用星火对这些数据做一些分析。我想要有容错能力，所以卡夫卡开始发挥作用。因此，我要做的是以下几点(如果我错了请纠正我)：
使用反应性流以固定的间隔从外部源获取数据。
把结果引向卡夫卡主题
使用火花流，为使用者创建流上下文
对消耗的数据执行分
浏览 3提问于2015-09-16得票数 4
回答已采纳
目前，在我的公司，我们将从Kafka 0.8迁移到0.11，代理迁移步骤，并在kafka文档中明确说明。
我陷入的是，升级卡夫卡客户端(生产者，消费者，火花流)，我没有发现任何文档/文章清楚地列出了哪些必要的更改或步骤来提升客户端，我所发现的只是java文档。
到目前为止，我所做的是将我的gradle中的kafka客户端版本更改为kafka-clients-0.11.0.0，从编译的角度看，一切都很好，没有任何代码更改。
我寻求帮助的是，除了卡夫卡-客户端版本之外，还有什么我应该解决的问题，客户端更改的指针吗？
浏览 2提问于2017-09-27得票数 2
回答已采纳
1回答
我在计算机的端口9999上创建了一个流。
我得在上面实现DGIM算法。然而，我无法逐个读取数据流中的位。下面是我的代码：
from pyspark import SparkContext
from pyspark.streaming import StreamingContext
import math
sc = SparkContext("local[2]", "NetworkWordCount")
ssc = StreamingContext(sc, 1)
当我使用以下命令时，我可以批量打印流：
lines.pprint()
ssc.start()
浏览 1提问于2018-07-18得票数 0
我有一个星团和一个卡夫卡星团。星星之火是从一个主题“一”中获取数据，处理它，并将其发送到同一个Kafka集群上的另一个名为“二”的主题。我使用kafka-console-producer.sh来编写主题"one“中的数据，使用kafka-console- and . am读取主题”2“中的数据。
这是我的火花代码：
# Read data from topic "one" and write it as it is in topic "two"
import sys
import os
from pyspark import SparkContext
浏览 0提问于2018-03-16得票数 0
2回答
我创造了一个卡夫卡的主题，并向它传递了一条信息。
所以
bin/kafka-console-consumer --bootstrap-server abc.xyz.com:9092 --topic myTopic --from-beginning --property print.key=true --property key.separator="-"
版画
key1-customer1
在命令行上。
我想从这个主题中创建一个卡夫卡流，并希望在控制台上打印这个key1-customer1。
我为此写了以下几封信：
final Properties streamsConfigu
浏览 9提问于2017-12-21得票数 9
回答已采纳
1回答
我有一个由卡夫卡流创造的数据。我想把它简化为一个值，然后在我的程序中使用这个值。
```scala
导入sparkSession.implicits._
val = sparkSession
.readStream
.format(“卡夫卡”)
.option("kafka.bootstrap.servers“)
.option(“订阅”，"theTopic")
.load()
val结果= df
.selectExpr(“将值转换为字符串)为json")
.map(json => getAnInt(Json)
.reduce { (x，y) =>
浏览 0提问于2019-07-20得票数 0
回答已采纳
我已经检查了合流卡夫卡流字数和异常检测的例子。在本例中，结果被写入主题。相反，如何通过REST或其他任何方法轻松、快速地将结果保存到远程数据库中。汇合平台上有什么结构吗？
代码示例: //而不是代码，发送远程数据库
wordCounts.toStream().to("streams-wordcount-output"，Produced.with(stringSerde，longSerde)；
浏览 0提问于2019-03-27得票数 1
回答已采纳
2回答
这里我使用pyspark做了一个简单的字数统计，我使用了cache方法，但似乎没有在第二次运行代码时加速：
from pyspark import SparkContext
from time import time
if __name__=='__main__':
sc=SparkContext()
rdd=sc.textFile("shakespear.txt")
rdd.cache()
t1=time()
rdd.flatMap(lambda r:r.split(" "))\
.map(
浏览 0提问于2017-07-23得票数 0
在我当前的场景中，Nifi收集数据，然后发送给Kafka。然后，任何流媒体引擎都会消耗卡夫卡的数据，并对其进行分析。在这种情况下，我不想在Nifi和流媒体引擎之间使用卡夫卡。因此，我想将数据从Nifi直接发送到流引擎。但是，我不知道一些细节。
例如，结构化流；如果我直接将数据从Nifi发送到Spark结构化流，Spark收到了这些数据，但是Spark的节点被关闭了。火花节点中的数据会发生什么变化？( Spark结构化流是否有任何Nifi接收器？)在这种情况下，Spark结构化流的数据保证是什么？
例如，；Storm有Nifi Bolt。但是，假设Storm已经收到了Nifi的数据，但是节点被关
浏览 0提问于2019-02-14得票数 0
回答已采纳
1回答
我试着和卡夫卡经纪人一起用日志。它不能集成在一起。
每个版本都是，
logstash 2.4
kafka 0.8.2.2与scala 2.10
kafka输入配置文件是：
input {
stdin{
}
}
output{
stdout {
codec => rubydebug
}
kafka {
bootstrap_servers => 10.120.16.202:6667,10.120.16.203:6667,10.120.16.204:6667'
topic_id => "ce
浏览 2提问于2016-10-09得票数 2
我对流媒体是个新手，在阅读文档后，我对此感到困惑。如果有新的东西到达，我不知道writestream是如何自动写入控制台的。它是在后台等待inputstream读取新内容，还是即使什么都没有到达，writestream仍然写入null？
//From doc
// Create DataFrame representing the stream of input lines from connection to localhost:9999
val lines = spark.readStream
.format("socket")
.option("host&
浏览 2提问于2019-09-15得票数 1
我需要帮助理解卡夫卡流行为时，max.task.idle.ms是使用卡夫卡2.2。
我有一个连接，其中KStream已被重新键入：
KStream stream1 = builder.stream("topic1", Consumed.with(myTimeExtractor));
KStream stream2 = builder.stream("topic2", Consumed.with(myTimeExtractor));
KTable table = stream1
.groupByKey()
.aggregate(myI
浏览 0提问于2019-05-21得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
