我正在尝试使用Spark streaming将csv文件从AWS S3读取到dataframe中，但是数据并没有存储在所需的列中，相反，它们只在1列中输入，而其他列是空的。需要一种方法，我可以采取的csv文件作为格式的输入。 我已尝试添加架构。删除架构并尝试推断架构状态时，必须指定架构。StructField("high",StringType,true) ::
StructField(&
浏览 12提问于2019-07-02得票数 0
回答已采纳
我目前正在使用从Kafka主题中读取json数据。json作为字符串存储在主题中。为了实现这一点，我提供了一个硬编码的JSON模式作为StructType。我正在寻找一种在流过程中动态推断主题模式的好方法。这是我的代码：(这是Kotlin，而不是通常使用的Scala) .readStream() .option
浏览 2提问于2020-10-21得票数 1
回答已采纳
13回答
我正在尝试将csv文件读取到数据帧中。我知道我的csv文件之后，我知道我的数据帧的模式应该是什么。另外，我使用spark csv包来读取该文件。我尝试指定模式，如下所示。databricks-datasets/wikipedia-datasets/data-001/pagecounts/sample
浏览 3提问于2016-10-08得票数 66
我正在使用pyarrow从s3读取拼图数据，我希望能够解析模式并将其转换为适合在Spark外部运行mLeap序列化模型的格式。 这需要解析模式。如果我有一个Pyspark数据帧，我可以这样做： test_df = spark.read.parquet(test_data_path)
schema = [ { "name" : field.simpleString读取数据<
浏览 47提问于2018-12-20得票数 0
(DataSource.scala:115) at org.apache.spark.sql.streaming.DataStreamReader.load(DataStreamReader.scala:232)
at org.apache.spark.sql.streaming</
浏览 18提问于2021-10-17得票数 4
回答已采纳
我已经在应用程序中创建了一个kafka流，并且可以解析任何通过它产生的文本。kafkaStream = KafkaUtils.createStream(ssc, zkQuorum, "spark-streaming-consumer", {topic: 1})
我想改变这一点，我对python还是新手，我如何更改流来解析avro消息？另外，在从Kafka读取Avro消息时，如何指定要使用的
浏览 4提问于2015-05-20得票数 4
我试图读取卡夫卡的数据，并将其上传到格林梅利的数据库使用火花。我使用的是格林梅-火花连接器，但我正在获取数据源，io.pivotal.greenplum.spark.GreenplumRelationProvider不支持流写入。是否格林梅源不支持流媒体数据？我曾尝试将数据源命名为“绿梅”，并将"io.pivotal.greenplum.spark.Gr
浏览 0提问于2019-04-04得票数 0
回答已采纳
对的引用
我认为我需要.options("inferSchema" , "true")和.option("header", "true")来打印我的标题，但显然我仍然可以打印我的csv的标题。它需要额外传递一次数据，默认情况下为false“的含义。
浏览 0提问于2019-07-08得票数 13
回答已采纳
我对spark结构化数据流和kafka的偏移管理是新的。使用spark-streaming-kafka-0-10-2.11。在消费者中，我如何从主题的特定分区中读取？applicationProperties.getProperty(BOOTSTRAP_SERVERS_CONFIG))
.option("subscribe", topicName) 我正
浏览 22提问于2019-05-29得票数 0
我能够从Kafka topic读取数据，并能够使用spark streaming在控制台上打印数据。spark = SparkSession \ .appName("StructuredSocketRead") \spark.spar
浏览 0提问于2020-11-03得票数 0
我正在尝试使用以下代码从Synapse的这个容器中读取数据： cfg = {"spark.cosmos.accountKey": accountKey,"spark.cosmos.container": con
浏览 58提问于2021-10-28得票数 0
回答已采纳
Spark streaming应用程序实时接收来自大量IoT设备的数据。但它们都是少量的数据。整体流程看起来像是-> Iot -> Kafka(1个主题/所有数据) ->火花流(过滤错误日志) -> DB(保存) ->警报屏幕 有没有使用spark或python进行实时日志分析的好方法？
浏览 13提问于2020-11-19得票数 0
回答已采纳
在spark批处理作业中，我通常会将JSON数据源写入到一个文件中，并且可以使用DataFrame阅读器的损坏列功能将损坏的数据写出到不同的位置，并使用另一个阅读器从同一作业中写入有效数据。(数据写为拼接) 但在Spark Structred Streaming中，我首先通过kafka将流作为字符串读取，然后使用from_jso
浏览 34提问于2018-12-27得票数 1
回答已采纳
我正在尝试将json消息从Kafka主题读取到PySpark数据帧中。Spark structured streaming或类似的方法来获得相同的数据帧。更新2:我设法添加了所需的包，并尝试从kafka读取消息：... .readStream \.. .lo
浏览 2提问于2021-06-04得票数 0
1回答
我已经编写了spark作业来读取一个文件，将数据转换为json并将数据发布到Kafka:我尝试了所有选项，如1. posting thread.sleep 2.更改linger.ms小于thread.sleep.But如果我写一个简单的独立制作人来发布消息到相同的kafka主题，它是没有任何问题的。因此，卡夫卡本身不存在任何问题。4.我可以看到我的发送方法是从日志中调用的，.Also在结束时被调用.No错误。以下是我的</em
浏览 0提问于2020-06-10得票数 0
1回答
我正在为一个用python编程的项目使用kafka和spark streaming。我想将kafka producer的数据发送到我的流媒体程序中。当我使用指定的依赖关系执行以下命令时，它可以顺利地工作：是否有任何方法可以指定依赖项并直接运行流代码(例如，不使用spark-submit或使用<e
浏览 2提问于2017-06-26得票数 0
1回答
使用Spark流的红移
、、、
我有一个Kafka - Spark流应用程序，可以每分钟摄取和处理60K个事件。我需要一个数据库来存储我转换后的数据帧，以供可视化层访问。Redshift可以与Spark Streaming一起使用吗?还是应该使用Cassandra？我将在每30秒的spark窗口中处理和存储数据帧。另外，我需要从每个窗口的数据存储中读取
浏览 0提问于2017-09-28得票数 0
2回答
那么我如何使用Spark2.0来读取这些嵌套的子文件夹，并从所有的叶json文件中创建一个静态的Dataframe呢？数据阅读器有“选项”吗？我的下一个目标是成为一个流DF，火龙将新文件持久化到s3中，使用Spark2.0中的新结构化流自然成为流数据的一部分。我知道这都是实验性的--希望有人以前曾使用过S3作为流文件源，在这里，
浏览 4提问于2016-10-30得票数 8
1回答
我正在使用Redis2.3.1从Kafka中读取一系列的值作为DStreams，我想对这些数据进行一些转换，比如取一个移动平均值，然后将其保存到PySpark中。streaming context (=connection to Spark)total = prices.reduce(lambda x, y: x +
浏览 32提问于2019-06-22得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
云直播活动推荐
腾讯云开发者
