我的spark streaming应用程序使用aws kenisis中的数据并部署在databricks中。我使用org.apache.spark.sql.Row.mkString方法来使用数据，整个数据都是以小写形式接收的。实际输入具有驼峰大小写字段名称和
浏览 10提问于2019-07-18得票数 0
回答已采纳
我试图读取卡夫卡的数据，并将其上传到格林梅利的数据库使用火花。我使用的是格林梅-火花连接器，但我正在获取数据源，io.pivotal.greenplum.spark.GreenplumRelationProvider不支持流写入。是否格林梅源不支持流媒体数据？我曾尝试将数据源命名为“绿梅”，并将"io.pivotal.greenplum.spark.GreenplumRe
浏览 0提问于2019-04-04得票数 0
回答已采纳
我有一个Spark结构的流媒体应用程序。应用程序从kafka接收数据，并且应该使用这些值作为参数来处理来自cassandra数据库的数据。我的问题是，我如何使用输入数据帧(kafka)中的数据，作为cassandra中的"where“参数"select”，而不会出现以下
浏览 25提问于2021-05-24得票数 0
回答已采纳
我们正在使用具有8核和32 We的Spark独立集群，具有相同配置的3个节点集群。
有时批处理流在不到1秒的时间内完成。在某些情况下，控制台中会出现以下日志，该时间超过10秒。
浏览 0提问于2016-03-29得票数 3
我正在阅读Spark2.2.0文档，发现这是对Static DataFrames 的奇怪引用
因为Spark2.0，DataFrames和数据集可以表示静态的、有界的数据，以及流的、无界的数据。与静态数据集/DataFrames类似，您可以使用公共入口点SparkSession (Scala/Java/Python/R )从流源创建流Dat
浏览 1提问于2017-12-24得票数 1
回答已采纳
2回答
星星之火2.1，结构化流与原始计数(*)，和(字段)是工作正常之上的地板文件，但过滤不起作用。), StructField(fieldname,StringType,true), StructField(valuestr,StringType,true))aggDF: org.apache.spark.sql.DataFrame =
浏览 0提问于2017-07-31得票数 6
回答已采纳
是否可以限制Kafka消费者返回Spark Streaming的批量大小？
我之所以问这个问题，是因为我得到的第一批记录有上亿条记录，需要很长时间才能处理和设置检查点。
浏览 0提问于2016-10-11得票数 18
回答已采纳
据我所知，Spark可以使用Spark Streaming分析流。Kafka可以从多个来源接收数据。我不明白的是，如果我有一个Kafka集群从多个来源接收数据，数据会被发送到一个运行Spark Streaming的数据库吗？或者Spark Streaming是否在应用服务器上运行？
浏览 5提问于2017-01-25得票数 0
回答已采纳
所以我在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.spar
浏览 3提问于2017-07-14得票数 5
回答已采纳
1回答
如何节流火花流？
、、
这个问题脱离了我关于管理AmazonDynamoDbClient节流和重试的另一个问题。然而，我认为解决方案可能存在之前，我甚至到发电机电话。我的高级过程如下:我有一个scala应用程序，它使用Apache读取大型CSV文件，并对它们执行一些聚合，然后将它们写入dynamo。我将其部署到EMR以提供可伸缩性。问题是，一旦聚合完成，我们就有数以百万计的记录准备进入发电机，但我们有一个写容量的发电机。它们不需要立即插入，但是最好能控制每秒多少次，这样我们就可以根据用例对其进行微调。下面是我到目前为止所得到<e
浏览 3提问于2017-03-28得票数 3
我有一个队列，我正在使用Spark Streaming从队列中读取数据。我需要将这些数据(在一些计算之后)写入2个不同的表(两个表的计算是不同的)。我注意到我的代码只执行第一个JavaPairDStream，而不执行另一个。我在两个PairStreams上都有输出操作。我还注意到，如果队列中的数据很小-比方说只有一条消息，那么它
浏览 0提问于2016-09-08得票数 0
我有一个Spark Structured Streaming Job，它从S3读取数据，转换数据，然后将其存储到一个S3接收器和一个Elasticsearch接收器。当这样做时，似乎Spark从S3源读取数据两次，每个接收器一次。
有没有一种更有效的方法来写入同一管道中的多个接收器？
浏览 1提问于2017-09-19得票数 14
我读过一些关于spark streaming的文章，我想知道是否有可能使用rabbitmq作为代理从自定义来源流式传输数据，并通过spark流提供这些数据，Spark的机器学习和图形处理算法将在这些数据上执行，并将其发送到其他文件系统/数据库/仪表板或客户接收器。附注:我用python编写代码，我没有任何使用spark
浏览 5提问于2020-09-02得票数 0
2回答
Kinesis将文件的持久性(在本例中为时间序列JSON )管理为一个文件夹层次结构，该层次结构由YYYY/MM/DD/HH (直到24小时编号)...great划分。那么我如何使用Spark2.0来读取这些嵌套的子文件夹，并从所有的叶json文件中创建一个静态的Dataframe呢？数据阅读器有“选项”吗？我的下一个目标是成为一个流DF，火龙将新文件
浏览 4提问于2016-10-30得票数 8
我想获取数据行，这是插入到外部mysql数据库后每2分钟。我想做这件事与火花流。但是在程序运行一次time.So后，我得到了这个错误，它第一次给了我数据，但在那之后，我得到了以下错误，程序终止了16/08/02 11:15:44 INFO JdbcRDD: closedat org.apache.spark.streaming.StreamingContext.start(StreamingContext.scala:594)
浏览 16提问于2016-08-03得票数 1
我正在使用kafka从远程mysql数据库读取数据，并将其传递到Pyspark streaming以应用情感分析，然后用该值更新mysql数据库中的列。这是最好的情况，还是有比这更有效的架构？
浏览 1提问于2019-11-20得票数 0
2回答
我的星火流应用程序是从卡夫卡读取使用DStream方法，我试图获得批量大小，以处理60,000条消息在10秒内。我所做的，
创建StreamingContext时，将批处理持续时间设置为1
浏览 0提问于2019-07-08得票数 2
回答已采纳
我试图编写简单的测试用例使用火花结构流。代码是受github上的启发的。的测试用例中运行它def demoCustomSink: Unit = { val doubled = input.toDS:43)
at org.apache.spark.sql.execution.streaming.StreamExecution.org$apache$spark<
浏览 0提问于2016-10-12得票数 1
1回答
无数据时停止提交作业
、、、
我使用spark streaming通过自定义接收器从mq中的队列中获取数据。Javastreaming上下文持续时间为10秒。并且为来自队列的输入定义了一个任务。在spark UI中的事件时间线中，我看到每隔10秒就会提交一个作业，即使没有来自接收方的数据
浏览 2提问于2018-03-15得票数 0
我想了解一下Spark Structured Streaming的单元测试方面。我的场景是，我从Kafka获取数据，并使用Spark Structured Streaming使用它，并在数据上应用一些转换。
我不确定如何使用Scala和Spark来测试它。有人能告诉我如何使用Scala在结构化流媒体中进行单元测试吗？我还是个流媒体新手。
浏览 3提问于2019-07-05得票数 2
云服务器
ICP备案
云直播
对象存储
实时音视频
腾讯云开发者
