在对数据执行聚合后，我使用pyspark将kafka流定向到redis。最终输出是一个流数据名。 .select(from_json(col("value").cast("string"), app_schema).alias("mob_apps"))
由于订阅到代理，这给了我流
浏览 15提问于2021-02-27得票数 0
Spark streaming应用程序实时接收来自大量IoT设备的数据。但它们都是少量的数据。整体流程看起来像是-> Iot -> Kafka(1个主题/所有数据) ->火花流(过滤错误日志) -> DB(保存) ->警报屏幕 有没有使用spark或python进行实时日志分析的好方法？
浏览 13提问于2020-11-19得票数 0
回答已采纳
我是在本地机器上使用Python的新手。到目前为止，我可以在Azure Databricks中编写代码。我想创建和部署连接到confluent kafka的库，并将数据保存到增量表。我弄糊涂了- 1]我是否需要通过如下设置从本地计算机连接到Databricks Delta，使用python将流存储到增量或将流存储到本地增量(我可以创建增量表
spark = pyspark.sql.
浏览 0提问于2021-01-08得票数 1
1回答
让我们假设我有一个具有表单的网站，用户可以在其中通过一些值。现在，我想获取这些值，使用Spark处理这些值，并将结果返回给用户。就像这样：详细的设置并不重要-星火流可以做一些建议或预测，可以坐在Databricks之上；后端可以是一个烧瓶应用程序.我的问题是：
这条管道漏掉了哪些碎片？一些中间DB，如Redis/Mongo/SQL？像卡夫卡这样
浏览 0提问于2018-06-16得票数 3
我很好奇，如果一个星火流应用程序被优雅地击落是绝对必须的，还是它有通过预写日志导致重复数据的风险。在下面的场景中，我概述了队列接收器与队列交互时需要确认消息的步骤序列。火花队列接收器将批消息存储到预写日志中。
预写日志中的消息通过流应用程序进行处理。火花队列接收器从队列中提取一批消息，这些消息已经
浏览 3提问于2015-07-29得票数 9
对于Databricks:转换数据帧并导出到xls / xlsx并保存到blob存储中是可能的吗？使用Python
浏览 41提问于2019-09-30得票数 0
我正在尝试将pandas dataframe的内容保存到windows/azure databricks的excel文件中。将熊猫作为pd导入 根据数据创建一个Pandas数据帧。writer = pd.ExcelWriter('pandas_simple.xlsx'，engine='xlsxwriter') 将数据帧</e
浏览 234提问于2020-07-03得票数 2
如何在Spark Structured Streaming中读取Kafka和查询外部存储，如Cassandra？我从Kafka获得消息流，我想对它应用Map操作，对于每个键，我想查询像Cassandra这样的数据存储，并获得该键的更多信息，并在流上应用进一步的操作。
浏览 2提问于2017-09-07得票数 1
1回答
我需要在从Kafka主题读取流之后，并在将其写入任何平台或表之前，对某些列应用函数。 .option("kafka.bootstrap.servers", bootStrapServers)
.option("subscribe", topicName)AS STRING)").withColumn("jsonData",from_json
浏览 25提问于2021-02-21得票数 0
回答已采纳
2回答
正如我从书中得到的，Kafka流状态存储是一种存储键/值存储，用于将数据存储到Kafka或过滤后。
为什么主题不能替代状态存储？
浏览 3提问于2020-06-27得票数 0
回答已采纳
我已经从scala中的数据集创建了CSV文件。它正在创建CSV文件，但是它是0字节的。当通过dbutils.fs.head(csv_path)读取时，其给出的错误 "java.lang.IllegalArgumentException:无法指向目录: csv_path“ ```result_dataset.coalesce(1) .write.format("com.databricks.spark.csv") .option(“字符集”，"UTF
浏览 35提问于2020-10-15得票数 0
1回答
我正在设计一个火花流应用与卡夫卡。我有以下几个问题:将RDBMS表中的数据流到kafka，并使用Spark使用者使用Spark处理消息和处理。问题: 1.我将数据从表流到kafka (键作为tablename，值作为表数据形式的JSON记录) --这是正确的体系结构吗？在spark使用者中，我试图使用DStream.foreachRDD(x =>转换
浏览 0提问于2017-01-31得票数 0
1回答
我正在使用Redis2.3.1从Kafka中读取一系列的值作为DStreams，我想对这些数据进行一些转换，比如取一个移动平均值，然后将其保存到PySpark中。说只有数据帧可以从PySpark写入。幸运的是，DStream在运行时会定期生成RDD -所以我必须弄清楚如何将RDD转换为Dataframe。我试过了 total.foreachRDD(lambda r
浏览 32提问于2019-06-22得票数 0
1回答
我有一个Azure系统，分为三个部分：
在databricks文件系统中安装存储后，需要处理一些数据。如何将位于databricks文件系统中</e
浏览 11提问于2020-11-09得票数 1
回答已采纳
我可以将数据帧保存到mongoDB，但是我在spark streaming中的程序提供了数据流( kafkaStream )，并且我无法将其保存在mongodb中，我也无法将此数据流转换为数据流。我们非常感谢您提供的任何意见。org.apache.spark.streaming.StreamingContext
import org.apache
浏览 0提问于2016-05-10得票数 0
现在，这两个对象都在Spark上运行，但我似乎不能确定为什么流不工作。 也许是redis在写入流端时实现它的接收器的方式，也可能是我尝试做这项工作的方式。我在流媒体上找到的几乎所有示例都与Spark samples有关，比如文本流媒体或TCP，而我在关系数据库上找到的唯一解决方案是基于Kafka connect的，我现在不能使用它，因为该公司没有在kafka我每天加载Oracle数据</em
浏览 21提问于2020-08-12得票数 2
1回答
获取示例形式的流数据
、、、
我正在尝试将一个函数(它与常规的火花数据处理)应用于流数据。在应用此函数之前，我需要对给定的数据使用.rdd.takeSample()，但这当然不适用于流数据。我使用以下结构化流代码获取流数据： .readStream \
.format("kafka") \.op
浏览 4提问于2020-04-02得票数 0
回答已采纳
1回答
我试着理解Connect给你买了什么流不给你的东西。我们有一个应用程序的一部分，我们想要使用一个主题并写到mariadb。为什么这是个坏主意？JDBC连接器给你买了什么？
浏览 0提问于2019-01-18得票数 11
回答已采纳
我使用的是spark 2.3.2，在对来自Kafka的2个或更多流媒体资源进行联合时遇到了一个问题。这些都是来自Kafka的流媒体源，我已经将它们转换并存储在Dataframes中。理想情况下，我希望将此UNIONed数据帧的结果以parquet格式存储在HDFS中，甚至可能存储回Kafka中。最终目标是以尽可能低的延迟存储这些合并的</
浏览 52提问于2019-07-02得票数 1
回答已采纳
1回答
我做了一个poc，在其中我使用火花流读取来自Kafka的数据。但是我们的组织要么使用Apache或者Kafka使用者来读取Apache中的数据作为一个标准过程。在我的应用程序用例中，我需要从kafka读取数据，过滤json数据，并在cassandra中放置字段，因此建议使用Kafka使用者，而不是flink/其他<em
浏览 4提问于2017-03-26得票数 0
云服务器
ICP备案
腾讯会议
云直播
对象存储
腾讯云开发者
