1回答
我收到Kafka发来的JSON字符串，需要PySpark处理。type":{"containsNull":true,"elementType":"double","type":"array"}}],"type":"struct"}\n^^^\n'
但是，如果我使用没有嵌套字段的模式现在，我可以获取除嵌套结构之外<
浏览 23提问于2020-05-11得票数 0
回答已采纳
2回答
我正在编写一个Kafka连接器，以便从Github上的多个来源下载一些数据(文本和yaml文件)，并将它们转换为某个类的对象，这些对象是从avsc文件自动生成的： {
"type": "record所以现在我有了一个对象图，我想把它保存在一个Kafka主题中。类型的值schema？: Invalid type for STRUCT: class MatomoRecord 有人能帮我定义基于avro模式
浏览 11提问于2019-06-04得票数 1
回答已采纳
我试图从Kafka向ClickHouse添加JSON数据。下面是简化的JSON： ..."sendCommChannelTypeCode":"SMS", ...下面是在ClickHouse中创建表的步骤，使用Kafka引擎创建另一个表，创建物化视图来连接这两个表，并将CH与Kafka</em
浏览 9提问于2020-12-02得票数 2
回答已采纳
1回答
我正在尝试使用结构化流媒体从Kafka中读取数据。从kafka收到的数据为json格式。我使用样例json创建模式，稍后在代码中使用from_json函数将json转换为数据帧以进行进一步处理。我面临的问题是嵌套模式和多值。示例模式将标记(比如a)定义为结构。从kafka读取的json
浏览 6提问于2019-06-11得票数 0
我已经开始研究apache，很少有关于apache的模式的查询。我想了解apache如何处理具有AVRO模式的Kafka主题(模式包括嵌套对象、对象数组等)因为我没有找到任何资源或例子，说明我们如何使用带有avro模式的Kafka注入数据。根据我的理解，apache必须为嵌套的Json对象提供平面模式或其他选项
浏览 1提问于2021-05-21得票数 0
我正在尝试使用结构化流从Kafka读取数据。从kafka收到的数据采用json格式。我的代码如下所示:在代码中，我使用from_json函数将json转换为dataframe以供进一步处理。"$app_version", StringType) .val df: DataFrame = spark.readStream
.format("kafka
浏览 3提问于2019-10-15得票数 4
回答已采纳
1回答
我正在尝试获得Kafka消息，并在独立的Spark处理它。Kafka将数据存储为json格式。我可以获取Kafka消息，但无法使用定义模式解析json数据。命令查看kafka主题中的kafka消息时，输出如下：
"{\"timestamp\":1553792312117,\"values\":[{\"id\":\"Sim
浏览 24提问于2019-03-26得票数 2
回答已采纳
我正在使用Kafka-Connect来实现一个Kafka-Elasticsearch连接器。生产者向Kafka主题发送了一个复杂的JSON，我的连接器代码将使用它来持久化Elastic search。连接器以结构()的形式获取数据。我可以在顶层Json中获取struct的字段值，但不能从嵌套的Json中获取。} },
"op"
浏览 0提问于2016-09-28得票数 4
1回答
我使用Spark3.2从Kafka 2.12-3.0.0获得JSON流。解析JSON后，我收到查询中的错误。Kafka主题流JSON：
b'{"pmu_id": 2, "time": 1642771653.06, "stream_id": 2,"analog": [], "digital": 0, "frequency16427346
浏览 5提问于2022-01-21得票数 1
回答已采纳
1回答
我有一个json文件，我想把它的内容发送给一个Kafka消费者。kafka使用Avro和一个遵守Json的模式我想发送.So有没有办法读取json，然后通过kafka发送它的全部内容，而不需要首先解析json，然后分别发送带有键和值的所有内容？ 谢谢。
浏览 17提问于2020-01-17得票数 1
我有一个使用PyFlink处理的事件流，其中事件是从AWS EventBridge中提取的。此流中的事件共享许多公共字段，但它们的detail字段根据source和/或detail-type字段的值而有所不同。，通过detail-type按下键输入不同的子表，其中每个子表的处理方式不同。我需要指定MAP的类型，例如MAP<VARCHAR, VARCHAR>，我不能使用MAP<>。
如何使用PyFlin
浏览 18提问于2022-04-10得票数 1
回答已采纳
在spark批处理作业中，我通常会将JSON数据源写入到一个文件中，并且可以使用DataFrame阅读器的损坏列功能将损坏的数据写出到不同的位置，并使用另一个阅读器从同一作业中写入有效数据。(数据写为拼接) 但在Spark Structred Streaming中，我首先通过kafka将流作为字符串读取，然后使用from_json获取我的DataFrame。然后，from_json使用JsonToStructs，它在解析</e
浏览 34提问于2018-12-27得票数 1
回答已采纳
我需要基于特定的JSON属性将消息从一个Kafka主题复制到另一个主题。也就是说，如果属性值是"A“-复制消息，否则不要复制。我正试图找出用KSQL最简单的方法来实现它。我的源消息都具有我的测试属性，但在其他方面有非常不同和复杂的模式。是否有一种方法可以对此设置“无模式”？CREATE OR REPLACE STREAM Test1 () kafka</e
浏览 7提问于2022-07-22得票数 0
我有一个具有嵌套字段和kafka_handle_error_mode='stream',input_format_import_nested_json=1设置的kafka_handle_error_mode='stream',input_format_import_nested_json=1表的Clickhouse服务器和两个物化视图：
_error='' case，将数据存储到与Engine=Kaf
浏览 12提问于2021-12-10得票数 0
我需要使用Kafka主题，它为每一行生成动态Json字符串，我无法在没有模式的情况下解析Json字符串。在我的例子中，Schema可以是动态的。这样我就可以使用接受DATASET作为输入的spark.read.json，来解析json的模式。val klines = spark.
readStream.format(
浏览 1提问于2020-06-04得票数 0
回答已采纳
我创建了一个带有JSON格式记录的Kafka主题。./kafka-console-consumer.sh --new-consumer \ --from-beginning \ --consumer.config /root/client.propert
浏览 5提问于2017-05-24得票数 1
回答已采纳
1回答
我用动态模式扼杀了Kafka spark streaming。我从卡夫卡(KafkaUtils.createDirectStream)消费每个消息/JSON字段可以嵌套，每个字段可以出现在一些消息中，有时不出现。我发现的唯一一件事就是：
案例类MyTyp(column1: OptionAny，column2: OptionAny....)这将包括，我不确定，可能出现的字段和嵌套文件。
浏览 10提问于2017-07-10得票数 0
我有一个火花结构化流scala作业，它从kafka读取json消息并将数据写入S3。我配置了一个合流模式注册中心，该模式使用type=object的json格式。现在，我可以从注册表中检索模式，但是我需要在包含kafka记录的dataframe上使用这个模式。如何将valueRestResponseSchema转换为结构化类型，以便能够在from_json中应用？”)
有可用<em
浏览 7提问于2022-11-30得票数 0
我们正在通过Kafka主题从上游系统接收Json消息。需求是以一定的间隔将这些消息存储到HDFS中。因为我们要存储到HDFS中，所以我们希望将这些记录中的某些数量合并到单个文件中。关于即将到来的唱片：
这些都是经过验证的消息，甚至NiFi处理器也能够解析它
浏览 1提问于2019-03-28得票数 2
我已经为记录创建了一个Avro模式，我们将其发布到我们的Kafka主题。我们实际的Kafka记录模式要复杂得多，但为了简洁起见，我只是附上了相关的部分。我们在记录中有多个嵌套的子类，但由于某些原因，我在尝试发布记录时得到以下异常(包名称已被隐藏)： java.lang.ClassCastException: aaa.bbb.ccc.ddd.Amount模式的当前子集。(KafkaRecord)的<e
浏览 20提问于2019-04-24得票数 0
云服务器
ICP备案
实时音视频
对象存储
即时通信 IM
腾讯云开发者
