我有一个可以收听Kafka主题的Spark流媒体应用程序。当获得数据时，我需要对其进行处理，并将其发送到Kudu。目前，我正在使用org.apache.kudu.spark.kudu.KuduContext API，并对数据框调用插入操作。为了从我的数据创建数据框，我需要调用collect()，以便可以使用sqlContext创建数据</
浏览 7提问于2018-08-08得票数 1
我使用impala(JDBC)两次来获取kafka偏移量并将数据保存在foreachRDD中。#node-1val messages = KafkaUtils.createDirectStream(*,newOffsets,)
messages.foreac
浏览 0提问于2018-03-15得票数 0
4回答
我想读一下卡夫卡的话题，然后用火花流的方式把它写到kudu表上。val parsed = sparkSession .format("kafka")kafka
at org.apache.<
浏览 51提问于2017-10-26得票数 2
回答已采纳
我有我的spark应用程序，从Kafka读取数据并摄取到Kudu。它已经成功运行了近25个小时，并将数据摄入到了Kudu中。在那之后，我看到从kafka日志中选出了新的kafka分区的领导者。我的应用程序进入完成状态，并出现以下错误
org.apache.spark.SparkException: ArrayBuffer(kafka.common.NotLeaderForParti
浏览 20提问于2018-02-13得票数 0
我有一个服务器，每1秒生成一些日志文件，我想使用Apache Spark处理这个文件。
我使用python编写了一个spark应用程序，并在while循环中处理了一组日志文件。我应该使用一个无限的while循环，还是应该在cron job甚至像airflow这样的调度框架中运行我的代码？
浏览 0提问于2017-04-18得票数 0
我使用的是星星之火-SQL2.4.1、星火-卡桑德拉-连接器_2.11-2.4.1.jar和java8。同时将数据从kafka主题插入到C*/Cassandra表数据。我正在犯错误：
org.apache.spark.sql.kafka010.KafkaDataConsumer - KafkaConsumer cache hitting max capacity of64, removing consum
浏览 0提问于2019-10-18得票数 0
Spark streaming应用程序实时接收来自大量IoT设备的数据。但它们都是少量的数据。整体流程看起来像是-> Iot -> Kafka(1个主题/所有数据) ->火花流(过滤错误日志) -> DB(保存) ->警报屏幕 有没有使用spark或python进行实时日志分析的好方法？
浏览 13提问于2020-11-19得票数 0
回答已采纳
我正在将一个Java项目从Cloudera 5.10升级到Cloudera 6.2。我们有Spark Streaming从Kafka读取数据来处理它并将结果写到其他地方。在升级过程中，Spark从v1.6升级到v2.1，Kafka从v0.8升级到v2.1。为了执行流处理，我们使用KafkaUtils.createStream(.
浏览 11提问于2021-09-27得票数 0
我使用Kafka 0.8.2从AdExchange接收数据，然后使用Spark Streaming 1.4.1将数据存储到MongoDB。我的问题是当我重新启动我的Spark Streaming作业时，比如更新新版本，修复bug，添加新功能。它将继续阅读最新的offset of kafka，然后我将失去数据AdX推动卡夫卡在重新
浏览 5提问于2015-08-06得票数 14
回答已采纳
2回答
当我们创建spark数据帧时，我们将数据帧中的数据发送到Kudu和Kafka(依次被提取并进入S3)现在，如果我使用相同的数据帧并将其转换为Kafka可以使用的格式
override def getKafkaDataFrame(df: DataFrame) : DataFra
浏览 51提问于2020-02-07得票数 0
我正在为一个使用MySQL作为其数据存储的应用程序构建分析功能。我们有一个基于微服务的架构，也使用Kafka。我们的用例并不真正需要“实时”分析，但这可能会在以后添加。对于我的用例，我想使用Tableau作为可视化平台，其中的报告将直接嵌入到web应用程序中。我的微服务将使用Avro模式注册表将事件推送到相关主题，然后报告微服务将使用这些事件并更新星型模式。
现在我的问题是:实现从Kafka
浏览 0提问于2018-03-23得票数 3
2回答
实时事件处理
、、
在服务器端，我想以这样的方式处理所有这些事件，即来自传感器的关于最新湿度、温度、压力...etc的信息将被存储/更新到数据库中。我真的需要spark，(flume/kafka) + spark，来满足处理端吗？我们可以在没有水槽的情况下使用flume进行任何类型的处理吗？
浏览 13提问于2017-06-29得票数 1
我已经设置了星火结构化流(Spark2.3.2)来阅读Kafka (2.0.0)。如果消息在星火流作业启动前进入主题，则无法从主题开始使用。这种预期的星火流行为是否忽略了最初运行Stream作业之前生成的Kafka消息(即使使用.option("stratingOffsets"，“最早”))？使用以下命令启动spark：spark-shell --packages org.apache.spark
浏览 1提问于2019-06-19得票数 8
回答已采纳
我对spark非常陌生，并从pyspark开始，我正在学习使用pyspark将数据从kafka推送到hive。SparkSessionfrom pyspark.sql.functions import *from os.path import abspath
warehou
浏览 2提问于2020-02-17得票数 0
我对大数据完全陌生，从最近几周开始，我一直在尝试构建日志分析应用程序。谁能建议几件事，如1)我如何实时读取服务器日志并将其传递给kafka broker。2)有没有从日志推送
浏览 1提问于2016-05-30得票数 0
我有一个练习，实现一个服务，从Kafka获取数据，处理它，并使用Spark Streaming将结果存储到elasticsearch中。我可以从Kafka获取数据到我的服务中，并在Spark集群中处理它，但我不知道如何在操作中将结果持久化到elasticsearch中。Process event t here /
浏览 0提问于2018-10-25得票数 0
试过：-从512到4096，较少失败，但即使是10s，失败仍然存在。.config("spark.streaming.kafka.maxRatePerPartition", "256")
.config(&quo
浏览 4提问于2017-02-16得票数 6
我正在分析日志，我有这样的架构：我的主要目标是在流媒体中创建机器学习模型。我认为我可以做两件事：2) Kafka -> spark Streaming-> elasticsearch -> sparkstreaming</
浏览 1提问于2017-05-10得票数 4
Spark从kafka获取数据。我想通过spark-streaming限制记录消耗。关于卡夫卡的数据非常庞大。我使用spark.streaming.kafka.maxRatePerPartition=1属性来限制spark中的记录。但是仍然在5分钟内，我收到了13400条消息。我的spark程序每5分钟不能处理超过1000条消息。Kafka to
浏览 17提问于2018-02-06得票数 0
据我所知，Spark可以使用Spark Streaming分析流。Kafka可以从多个来源接收数据。我不明白的是，如果我有一个Kafka集群从多个来源接收数据，数据会被发送到一个运行Spark Streaming的数据库吗？或者Spark Streaming是否在应用服务器上运行？
浏览 5提问于2017-01-25得票数 0
回答已采纳
云服务器
ICP备案
云直播
腾讯会议
实时音视频
腾讯云开发者
