腾讯云
开发者社区
文档
建议反馈
控制台
首页
学习
活动
专区
工具
TVP
最新优惠活动
文章/答案/技术大牛
搜索
搜索
关闭
发布
登录/注册
精选内容/技术社群/优惠产品,
尽在小程序
立即前往
文章
问答
(9999+)
视频
沙龙
1
回答
将
kafka
主
题中
的
数据
读入
spark
dataframe
apache-spark
、
apache-spark-sql
、
spark-streaming
private static final SparkSession sparkSession = SparkSession.builder().master("local[*]").appName("
Spark
2JdbcDs").option("
kafka
.bootstrap.servers", "localhost:9092") .option("subscribe", "SqlMessages").load();
浏览 20
提问于2021-01-15
得票数 0
1
回答
多个
kafka
集群
的
Spark
结构化流
apache-spark
我正在尝试从不同集群
的
多个主
题中
获取一些关于
spark
结构化流媒体和阅读
的
基本信息。我找到了一些教程,这些教程展示了如何从同一集群上
的
多个
kafka
主
题中
拉出
数据
,但如果这些主题位于不同
的
集群中呢?如何
将
多个
kafka
集群中
的
主题
读入
spark
?
浏览 14
提问于2018-07-18
得票数 0
回答已采纳
1
回答
将
星火流PySpark
数据
格式写入Cassandra覆盖表而不是追加
apache-spark
、
apache-kafka
、
cassandra
、
spark-structured-streaming
、
spark-cassandra-connector
我在运行一个由卡夫卡,斯派克和卡桑德拉组成
的
单节点集群。都在同一台机器上。 从一个简单
的
Python脚本中,我每5秒
将
一些虚拟
数据
流到
Kafka
主
题中
。然后,使用
Spark
结构化流,我
将
这个
数据
流(每次一行)
读入
PySpark
DataFrame
中,其中包含startingOffset = latest。最后,我试图
将
这一行附加到已经存在
的
Cassandra表中。
浏览 1
提问于2020-10-21
得票数 0
回答已采纳
1
回答
Pyspark
Kafka
结构化流:写出时出错
apache-spark
、
pyspark
、
apache-kafka
、
spark-structured-streaming
我能够在PySpark中通过两个不同
的
步骤从
Kafka
主题读取流,并将(转换后
的
)
数据
写回另一个
Kafka
主题。执行此操作
的
代码如下所示:df =
spark
\ .format("
kafka
") \ .option("topic",
浏览 0
提问于2018-10-07
得票数 0
1
回答
Spark
kafka
streaming -如何确定批次结束
apache-spark
我正在使用
Kafka
流媒体从
Kafka
主题消费。(
kafka
直播流)该topic中
的
数据
每隔5分钟就会从其他来源到达。现在我需要处理每隔5分钟到达
的
数据
,并将其转换为
Spark
DataFrame
。现在,流是连续
的
数据
流。我
的
问题是,我如何确定我已经读完了加载到
Kafka
主
题中
的
第一组
数据
?(这样我就可以将其转
浏览 0
提问于2017-06-09
得票数 0
1
回答
如何
将
spark
流输出包装在数组括号中?
scala
、
apache-spark
、
spark-structured-streaming
Spark
:2.3.0 Scala:2.11.12输入
数据
[{"foo":"bar"}] 当我使用select(from_json(col(A), mySchema
浏览 1
提问于2018-08-08
得票数 1
回答已采纳
1
回答
使用python火花向
Kafka
发送大型CSV
python
、
apache-spark
、
apache-kafka
、
pyspark
、
kafka-python
我正试着给卡夫卡送一辆大
的
CSV。基本结构是读取CSV
的
一行并使用标头将其压缩。from
kafka
import SimpleProducer, KafkaClientproducer = SimpleProducer(
kafka
)使用PYSPARK,我轻松地从CSV文件创建了一个消息
的
浏览 4
提问于2015-08-31
得票数 6
回答已采纳
2
回答
如何显示一个流
DataFrame
(AnalysisException显示失败)?
apache-spark
、
pyspark
、
apache-kafka
、
spark-structured-streaming
所以我在
Kafka
主
题中
有一些
数据
,我把这些流
数据
放到一个
DataFrame
中。我想要显示
DataFrame
中
的
数据
:from
kafka
import KafkaProducer from pyspark.sql import SparkSession,
DataFrame
:
spark
-sql-
kafka
-0-10_2.11:2.2.0,org.apache
浏览 3
提问于2017-07-14
得票数 5
回答已采纳
1
回答
以
Kafka
为源
的
结构化流中
的
JSON模式推理
apache-spark
、
apache-kafka
我目前正在使用从
Kafka
主
题中
读取json
数据
。json作为字符串存储在主
题中
。为了实现这一点,我提供了一个硬编码
的
JSON模式作为StructType。这是我
的
代码:(这是Kotlin,而不是通常使用
的
Scala) .readStream() .option("
kafka
.bootstrap.servers.tri
浏览 2
提问于2020-10-21
得票数 1
回答已采纳
2
回答
从
Kafka
主
题中
读取消息并将其转储到HDFS中
java
、
scala
、
apache-spark
、
apache-kafka
、
hdfs
我正在尝试使用
Kafka
主
题中
的
数据
,将其加载到Dataset中,然后在加载到Hdfs之前执行过滤。我可以从
kafka
主
题中
消费,将其加载到dataset中并保存为HDFS中
的
拼图文件,但无法执行过滤条件。你能分享一下在保存到hdfs之前进行过滤
的
方法吗?我正在使用Java和
Spark
来消费来自
kafka
的
主题。我
的
部分代码是这样
的
: DataframeDeser
浏览 6
提问于2020-05-05
得票数 1
1
回答
writeStream:在一个单独
的
json文件中
的
每个
数据
帧行
pyspark
、
spark-streaming
、
pyspark-dataframes
我正在使用pyspark
将
Kafka
主
题中
的
数据
作为流
数据
读取,如下所示: .appName("
Spark
Structured Streamingfrom
Kafka
") \ .readStream \ .format("
kafka
浏览 2
提问于2020-03-12
得票数 0
回答已采纳
1
回答
通过火花流或水槽
将
Xml转换为Avro,从
Kafka
转换为hdfs
apache-spark
、
apache-kafka
、
spark-streaming
、
avro
、
spark-avro
数据
将
采用xml格式,并将首先触及
kafka
主题。然后,我可以使用水槽或火花流吞食和转换从xml到avro,并在hdfs中
的
文件。我有cloudera环境。当avro文件命中hdfs时,我希望以后能够将它们
读入
蜂窝表。 我想知道做这件事最好
的
方法是什么?我已经尝试过自动模式转换,比如星星之火-avro(这没有火花流),但是问题是
spark
avro转换
数据
,但是hive不能读取它。
Spark
将
xml转换为
da
浏览 1
提问于2016-05-31
得票数 0
回答已采纳
3
回答
如何
将
数据
帧转换为JSON,并使用key写入
kafka
主题
scala
、
dataframe
、
apache-spark
、
apache-spark-sql
、
spark-streaming-kafka
我正在尝试以JSON格式向
kafka
写入
数据
帧,并在Scala中向
数据
帧添加密钥,我目前正在使用来自
kafka
-
spark
的
示例: .write .option("
kafka
.boots
浏览 0
提问于2019-07-30
得票数 2
1
回答
Spark
Job未将消息发布到
Kafka
主题
scala
、
apache-spark
、
apache-kafka
、
spark-streaming
、
kafka-producer-api
我已经编写了
spark
作业来读取一个文件,
将
数据
转换为json并将
数据
发布到
Kafka
:我尝试了所有选项,如1. posting thread.sleep 2.更改linger.ms小于thread.sleep.But如果我写一个简单
的
独立制作人来发布消息到相同
的
kafka
主题,它是没有任何问题
的
。因此,卡夫卡本身不存在任何问题。4.我可以看到我
的
发送方法是从日志中调用
的
,.Also在结束时被调用.
浏览 0
提问于2020-06-10
得票数 0
1
回答
Spark
Structured streaming: JDBC接收器中
的
主键
mysql
、
apache-spark
、
apache-spark-sql
、
spark-structured-streaming
、
apache-spark-dataset
我正在使用带有更新模式
的
结构化
数据
流从
kafka
主
题中
读取
数据
流,然后进行一些转换。 然后,我创建了一个jdbc接收器来使用Append模式
将
数据
推送到mysql接收器中。问题是我如何告诉我
的
接收器让它知道这是我
的
主键,并基于它进行更新,这样我
的
表就不应该有任何重复
的
行。val df:
DataFrame
=
spark
.form
浏览 6
提问于2019-05-02
得票数 1
回答已采纳
1
回答
如何以编程方式
将
Kafka
主题加载和流到PySpark
数据
python
、
pyspark
、
apache-kafka
、
kafka-consumer-api
卡夫卡有许多读/写火花
数据
的
方法。我试着阅读来自卡夫卡主题
的
信息,并从中创建一个
数据
框架。可以从主
题中
提取消息,但无法将其转换为
数据
文件。任何建议都会有帮助。message df = data.map # am unable to convert it to a
dataframe
我也试过在下面, .read
浏览 1
提问于2020-06-12
得票数 1
1
回答
如何显示流查询中间转换
的
结果?
scala
、
apache-spark
、
spark-structured-streaming
我正在实现一个用例来试用
Spark
Structured Streaming API。源
数据
是从
Kafka
主
题中
读取
的
,经过一些转换后,结果会写入控制台。我希望打印中间输出以及结构化流查询
的
最终结果。以下是代码片段:
浏览 4
提问于2019-07-31
得票数 1
回答已采纳
1
回答
spark
流式传输到pyspark json文件中
的
数据
帧
python
、
json
、
pyspark
、
spark-streaming
我需要火种
的
帮助。我正在从
kafka
流式传输json
数据
,我需要转换为pyspark中
的
Dataframe
。为了流媒体,我使用了下面的代码。("
spark
.executor.instances", 4) \ .getOrCreate()value)) lines.foreachRD
浏览 3
提问于2017-07-30
得票数 0
2
回答
如何使用flink对两个
kafka
主题进行
数据
比较
java
、
apache-kafka
、
comparison
、
apache-flink
我被困在这样
的
场景中,我需要使用来自2个或更多
kafka
主题
的
数据
流(在应用过滤器之后)并对
数据
进行比较。由于
数据
量很大,请提出现有的最佳选择。弗林克有可能吗?
浏览 4
提问于2020-02-04
得票数 0
回答已采纳
2
回答
使用pyspark批处理作业读取
Kafka
的
最新记录
apache-spark
、
apache-kafka
我正在pyspark中执行一个批处理作业,其中
spark
将
每5分钟读取一次
kafka
主
题中
的
数据
。df =
spark
\ .format("
kafka
") \ .test") \ .option("startingO
浏览 2
提问于2020-03-04
得票数 0
点击加载更多
扫码
添加站长 进交流群
领取专属
10元无门槛券
手把手带您无忧上云
相关
资讯
Spark 2.4 原生支持了avro,别再引用第三方库了
如何将外部数据源与Spark集成?
Apache Spark 2.4 内置的 Avro 数据源实战
大数据有道之spark选择去重
实时计算框架 Flink 在教育行业的应用实践
热门
标签
更多标签
云服务器
ICP备案
实时音视频
即时通信 IM
对象存储
活动推荐
运营活动
广告
关闭
领券