如何自动选择文件作为kafka制作人的输入？
在Kafka中，可以通过使用Kafka Connect来实现自动选择文件作为制作人的输入。Kafka Connect是Kafka的一个工具，用于将外部系统与Kafka进行连接和集成。
要自动选择文件作为Kafka制作人的输入，可以按照以下步骤进行操作：
- 配置Kafka Connect：首先，需要在Kafka Connect的配置文件中指定输入源为文件。可以通过编辑配置文件
connect-standalone.properties或connect-distributed.properties来完成。在配置文件中，需要设置以下属性：connector.class：指定使用的连接器类，通常为FileStreamSource。file：指定要读取的文件路径。- 其他可选属性：根据需要，可以设置一些可选属性，如文件解析器、编码方式等。
- 创建连接器配置文件：接下来，需要创建一个连接器配置文件，用于指定Kafka主题、分区、数据格式等信息。可以创建一个JSON或者properties格式的文件，并在其中设置以下属性：
name：连接器的名称。connector.class：连接器类，同样为FileStreamSource。tasks.max：指定连接器的最大任务数。topic：指定要写入的Kafka主题。- 其他可选属性：根据需要，可以设置一些可选属性，如数据格式、分区策略等。
- 启动Kafka Connect：在配置好Kafka Connect和连接器配置文件后，可以启动Kafka Connect来开始文件的自动选择和传输。可以使用以下命令启动Kafka Connect：
- 启动Kafka Connect：在配置好Kafka Connect和连接器配置文件后，可以启动Kafka Connect来开始文件的自动选择和传输。可以使用以下命令启动Kafka Connect：
- 其中，
config/connect-standalone.properties是Kafka Connect的配置文件路径，connector-config-file.properties是连接器配置文件路径。 - 监控和管理：一旦Kafka Connect启动并开始读取文件并将数据写入Kafka主题，您可以使用Kafka工具或命令行来监控和管理数据流。可以使用Kafka消费者来消费Kafka主题中的数据，并使用Kafka管理工具来管理连接器和任务。
总结：通过配置Kafka Connect和连接器配置文件，可以实现自动选择文件作为Kafka制作人的输入。这样，文件中的数据将被读取并传输到指定的Kafka主题中，以供后续处理和分析。腾讯云提供了Kafka服务，您可以使用腾讯云的Tencent Kafka来搭建和管理Kafka集群，并使用Kafka Connect来实现文件到Kafka的自动传输。
相关·内容
3回答
我正在读取一个csv文件，并将该输入的行提供给我的Kafka制作人。现在我要我的卡夫卡制作人以每秒100条的速度产生信息。
浏览 3提问于2017-07-10得票数 3
回答已采纳
2回答
现在，我的kafka生成器正在将所有消息下沉到kafka主题的单个分区中，该主题实际上具有超过1个分区。
如何创建一个生产者，它将使用默认分区程序并在主题的不同分区之间分发消息。
我的kafka制作人的代码片段：
Properties props = new Properties();
props.put(ProducerConfig.RETRIES_CONFIG, 0);
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,bootstrap.servers);
props.put(ProducerConfig.ACKS_CONFIG,
浏览 0提问于2018-08-28得票数 0
我正在使用SpringCloudStream3.2.2为kafka主题生成云事件消息。根据Cloud规范，在使用kafka时，标头应该使用"ce_“前缀。下面是我的制作人的样子：
this.streamBridge.send(
"myBinding-out-0",
CloudEventMessageBuilder.withData(event)
.setSource("my-source")
.setId(UUID.randomUUID().toString())
浏览 7提问于2022-04-11得票数 0
嗨，我正在浏览，我遇到了一个错误，当我输入bin/kafka-console-producer.sh --broker-list localhost:9093 --topic first时
当我期待一个文本输入将消息发送给Kafka制作人时，我什么也得不到。我已经按照单代理和多代理设置的说明进行了操作，但似乎什么都不起作用。我的Zookeeper正在运行，我看到在正确的日志目录中创建了日志文件，但没有关于我的消息。
我也尝试过像so cat test.txt | bin/kafka-console-producer.sh --broker-list localhost:9093 --topic
浏览 0提问于2017-09-18得票数 3
我是个骆驼新手，正在尝试一些代码来理解MockEndpoints功能是如何工作的，但它不起作用。这里是我的骆驼路线，我想用MockEndpoints进行单元测试。属性值在application.properties文件中定义。
from("kafka:{{kafka.servers}}?topic={{kafka.consumer.topic}}&groupId={{kafka.consumer.groupId}}")
.to("direct:notify");
from("direct:notify")
.log(&#
浏览 3提问于2017-07-21得票数 3
回答已采纳
1回答
假设我的制作人正在制作一些消息，而Kafka消费者正在使用这些消息。现在我要把这些信息发送出去。我如何用python将卡夫卡与火花连接起来，并将卡夫卡的信息发送给火花呢？
浏览 4提问于2022-05-12得票数 -2
1回答
我有一个卡夫卡主题的制作人，它一直在向卡夫卡主题推送一些信息。此外，我还有另一个服务在阅读来自主题的这些消息。
我有一个业务用例，有时使用者需要忽略队列中已经存在的所有消息，并且只开始处理新的即将出现的消息。可以在不停止和重新启动kafka服务器的情况下对其进行归档。
我正在忙着走。因此，如果kafka支持这样的要求，我是否可以改变消费者的配置，使用sarama GO客户端从最新消息开始消费。
提前谢谢你。
浏览 9提问于2022-09-15得票数 0
在Apache中的分区主题上，我有点困惑。因此，我正在绘制一个简单的用例，我想知道在不同的场景中会发生什么。所以这里是：
我有一个主题T，它有4个分区( TP1、TP2、TP4和TP4 )。
假设我有8条消息( M1 to M8 )。现在，当我的制作人将这些消息发送到主题T时，Kafka broker将如何在以下情况下接收这些消息：
场景1:只有一个kafka broker实例具有前面提到的分区的主题T。
场景2:有两个kafka broker实例，每个节点具有相同的主题T和前面提到的分区。
现在假设kafka broker实例1下降了，消费者将如何反应？我假设我的消费者是从broker实例1
浏览 5提问于2016-01-13得票数 0
回答已采纳
2回答
制作人通过设置卡夫卡经纪人列表发送信息，如下所示。
props.put("bootstrap.servers", "127.0.0.1:9092,127.0.0.1:9092,127.0.0.1:9092");
我想知道“制片人”如何知道三个经纪人中哪一个知道哪一个有分区领导。对于典型的分布式服务器，要么有承载服务器，要么有虚拟IP，但是对于Kafka，它是如何加载的？生产者程序是否试图随机连接到一个代理，并寻找具有分区领导的代理？
浏览 0提问于2018-03-23得票数 5
回答已采纳
我有以下制作人代码：
var kafka = require('kafka-node');
var KafkaProducer = kafka.Producer;
var KeyedMessage = kafka.KeyedMessage;
var jsonRequest = JSON.stringify(request.object);
//I have to define client every time this endpoint is hit.
var client = new kafka.Client();
var producerKafka = new Kaf
浏览 0提问于2018-05-16得票数 0
我刚刚开始使用Apache /，并且遇到了一些问题，试图在AWS上设置一个集群。目前我有三台服务器：
一个跑动物园管理员两个跑卡夫卡。
我可以在没有问题的情况下启动Kafka服务器，并且可以在这两个服务器上创建主题。然而，当我试图在一台机器上启动一台生产商，在另一台机器上启动一个消费者时，问题就来了：
卡夫卡制作人：
kafka-console-producer.sh --broker-list <kafka server 1 aws public dns>:9092,<kafka server 2 aws public dns>:9092 --topic samsa
浏览 3提问于2015-12-14得票数 8
回答已采纳
我是Kafka.When中使用kafka 0.9.0的新手，我可以通过设置我的卡夫卡制作人的compression.type属性来设置编解码器。
假设我在我的制作人中使用snappy压缩，当我使用某种卡夫卡-消费者来消费来自卡夫卡的信息时，我应该做些什么来解码来自快速的数据，还是它是卡夫卡消费者的内置特性呢？
在office中，我无法在kafka使用者中找到与编码相关的任何属性(只涉及生产者)。
有人能澄清这件事吗？
浏览 0提问于2016-05-25得票数 0
我有一个使用confluent- kafka -go绑定的Go编写的kafka制作人。 我希望我的kafka生产者建立到kafka代理的连接，该代理位于支持HTTP连接隧道的HTTP代理之后。 librdkafka支持这一点吗？confluent-kafka-go支持这一点吗？如果是，请告诉我相关的文档，或者提供一个代码片段来说明如何做到这一点。 如果不支持，添加支持会有多难？
浏览 21提问于2021-09-30得票数 0
我有十个Kafka制作人，每个人都写一个主题的不同分区。 我不知道哪个更有效。 让一个消费者监听十个分区，还是让十个消费者监听不同的分区？
浏览 64提问于2019-03-21得票数 0
1回答
我注意到有几个制作人：
org.apache.kafka.clients.producer.KafkaProducer<K,V>
kafka.javaapi.producer.Producer
它们之间的区别是什么？应该使用哪一个?为什么?请建议！
浏览 3提问于2016-10-14得票数 0
2回答
我是卡夫卡的新手，我试图在卡夫卡中建立一个消费者，让它阅读卡夫卡制作人发布的信息。纠正我如果我错了，我理解卡夫卡消费商店是否抵消了ZooKeeper？然而，我没有一个动物园管理员实例运行，并希望轮询，让我们说每5分钟看看是否有任何新的消息发布。
到目前为止，我的代码是：
import logging
from django.conf import settings
import kafka
import sys
import json
bootstrap_servers = ['localhost:8080']
topicName = 'test-info'
浏览 5提问于2020-04-07得票数 2
回答已采纳
我已经使用createDirectStream实现了星火流。我的Kafka制作人每秒钟发送几条消息给一个有两个分区的主题。
在星火流方面，我每秒钟都会看到卡夫卡的信息，我会在5秒的窗口大小和频率上加窗。
卡夫卡的信息被正确处理了，我看到了正确的计算和打印。
但是在Spark中，在流部分中，它将每个窗口的事件数显示为Zero。请看这张图片：
我很困惑，为什么它显示的是零，它不应该显示大量的卡夫卡信息被输入火花流？
更新：
当我使用groupByKeyAndWindow() api时，这个问题似乎正在发生。当我从代码中注释掉这个api使用时，Streaming开始正确地报告Kafka事件
浏览 1提问于2016-05-06得票数 3
1回答
卡夫卡输出流
、、
我已经和kafka消费者和生产者api合作了一段时间了，我想尝试一下streams api。我在网上看了很多推荐信，但我想不出这件简单的事情。
如何使只向输出主题发送消息的KStream。
例如，他们在github上有一个最基本的例子：，它从一个队列中获取消息，并在操作它之后将它们发布到另一个队列中。
就像这样：
final KStreamBuilder builder = new KStreamBuilder();
final KStream<String, String> textLines = builder.stream();
// do the dirty work...
浏览 2提问于2017-04-12得票数 1
回答已采纳
你好(我使用谷歌翻译)。我有以下问题，我在kubernetes有一个Kafka服务，一切都是由牧场主管理的，卡夫卡的部署是通过牧场主允许的目录来完成的，(我附上服务的图像)。
库伯奈特内部的一切都是正确的，但是现在我需要一个库伯奈特外部的制作人连接到卡夫卡并发送信息，以便在库伯奈特内部接收到这些信息。我未能完成这项任务，我已经尝试过按照本指南进行另一次kafka部署：
1
但是，我无法理解在牧场主目录的版本中，而不是在通过YAML文件安装的版本中，我应该在哪里以及如何配置一个在kubernetes之外的生产者，我还试图将服务设置为NodePort，但这是行不通的，欢迎您提供任何帮助
浏览 2提问于2022-08-14得票数 -1
2回答
我正在使用访问卡夫卡。我试着创造一个卡夫卡制作人：
kafka_producer = KafkaProducer(bootstrap_servers=['kafka:9092'])
但这在kafka.errors.NoBrokersAvailable: NoBrokersAvailable例外情况下失败。
我发现需要将api_version参数添加到KafkaProducer中：
kafka_producer = KafkaProducer(bootstrap_servers=['kafka:9092'],
浏览 0提问于2019-07-17得票数 10
1回答
这是我使用kafkajs的制作人。当我在options.username中提供了不正确的用户名时，我看到消息Connected。如何使用kafkajs处理连接错误 const { Kafka, logLevel } = require('kafkajs')
async function kafkaProducer(options) {
const kafka = new Kafka({
brokers: [options.bootstrapServer],
clientId: options.clientId,
ssl
浏览 50提问于2020-05-04得票数 0
1回答
我正在遵循本指南中的步骤
我得到的错误信息是
BadRequestException:连接器配置{.}不包含连接器类型
我在运行命令时
sh kafka_2.12-2.3.0/bin/connect-standalone.sh connect-standalone.properties snowflake_kafka_config.json
我的配置文件是
connect-standalone.properties
bootstrap.servers=localhost:9092
value.converter=org.apache.kafka.connect.json.JsonConvert
浏览 0提问于2019-10-15得票数 0
回答已采纳
4回答
目前，我正在从本地wlan接口中嗅探数据包，例如：
sudo tshark > sampleData.pcap
然而，我需要把这些数据提供给卡夫卡。
目前，我有一个卡夫卡制作人脚本producer.sh。
../bin/kafka-console-producer.sh --broker-list localhost:9092 --topic 'spark-kafka'
然后像这样向kafka提供数据：
producer.sh < sampleData.pcap
在sampleData.pcap中，我有预先捕获的IP数据包信息。
但是，我想让这个过程自动化，就像这样
浏览 11提问于2016-03-08得票数 3
1回答
卡夫卡是否提供了对其主题发布的信息进行操作的任何设施？我需要向我的Kafka代理发布大量的消息，但我更愿意将它们作为一个批处理捆绑在一起，然后(如果可能的话)让Kafka将它们分成单独的(较小的)记录/消息供me...and，然后将它们发布到一个主题中。
基本上，我现在的内容是：
制作人获取大量的信息并将其分割成单独的记录。
个人记录被发布到卡夫卡主题。
消费者消费卡夫卡的个人记录
作为所希望的行为，我需要的是：
制片人拿出一大块文本并将其发送给卡夫卡主题。
Kafka (不知怎么地)是用分裂逻辑来配置/编程的，它将这个blob分割成单独的记录。
消费者消费卡夫
浏览 4提问于2016-08-15得票数 1
回答已采纳
运行合流- Kafka 3.3.1e Kafka 0.11.0x
我有一个副本= 3的分区。
我的制作人正在运行ack=-1
该分区有一个不同步的副本。
复制滞后时间最大ms = 10000
最小insync副本=2
1)当ack=-1时，生产者会提交记录吗？
2)如何使不同步的副本恢复同步？
浏览 0提问于2019-03-20得票数 0
回答已采纳
使用循环调用kakfa produce类时，无法将消息写入kafka topic (producer)。 我对Python和Kafka非常陌生。我正在尝试编写一个python程序，将消息写入Kafka主题并生成，以便Kafka消费者可以订阅该主题以发布消息。 我不确定我的程序中缺少了什么，它限制了向主题写入消息。 注意:我正在读取一个JSON文件，并使用for循环准备键值。然后将其赋给一个变量，并将该变量与使用arg生成的Kafka引用为msg。 附件是卡夫卡制作人程序。 输入: Json_smpl.json 文件内容： {
"transaction":{
"Accn
浏览 36提问于2019-01-12得票数 0
我在Kotlin lambda里面运行一个Smallrye Kafka制作人。问题是，我大部分时间都会得到一个TimeoutException，我试图向Kafka发送消息(运行在艾文)。我读到，这可能是生产者的超时配置中的一个问题，但我想了解更多关于如何尝试和配置生产者的request.timeout.ms的信息，因为关于如何为小黑眼连接器这样做的信息非常有限。
浏览 2提问于2022-11-07得票数 0
在同一个Kafka/Spark流项目中，我已经导出了两个应用程序的两个罐子。卡夫卡制作人的罐子工作得很好。星火消费者的jar将返回以下错误：
NoClassDefFoundError: kafka/序列化程序/字符串解码器
它们共享相同的依赖文件夹，显然，为了使jars在Eclipse之外工作，我已经导出了该文件夹。
在Eclipse中，它们都工作得很好。
我怎么才能解决这个问题？
浏览 2提问于2017-06-15得票数 1
回答已采纳
2回答
你好，我在研究卡夫卡。我在试着理解卡夫卡的基本知识。我正在学习卡夫卡。我用码头安装了卡夫卡。目前我有一个经纪人。我使用以下命令创建了带有3个分区的主题。
kafka-topics --create --zookeeper zookeeper:2181 --replication-factor 1 --partitions 3 --topic topic2
之后，我创建了如下所示的制作人。
kafka-console-producer --broker-list localhost:9092 --topic topic2
>This is my producer
我在这里完全糊涂了。当
浏览 2提问于2019-11-15得票数 1
回答已采纳
我是三叉树的新手。我正在写一个从kafka读取数据的三叉树拓扑。主题名称为'test‘。我有本地的kafka设置。我在当地创办了zookeeper kafka。并用kafka创建了一个主题'test‘，打开制作人并输入消息'Hello Kafka!’。
我想用三叉树阅读“测试”主题中的消息“Hello Kafka”。
下面是我的代码。我得到了空的元组。
TridentTopology topology = new TridentTopology();
BrokerHosts brokerHosts = new ZkHosts("localhost
浏览 5提问于2015-09-12得票数 1
1回答
是在年代ms还是在几秒钟内？是否代表了卡夫卡收到的信息？还是卡夫卡-客户-制作人寄来的？或者它是什么时候消费的？，我如何知道我的VM与融合云中的Kafka VM之间的时钟时间的差异？
浏览 3提问于2021-05-30得票数 0
回答已采纳
3回答
我需要实现卡夫卡消费者在火花流的火花1.0。我写过卡夫卡制作人。有谁能帮我写一个火花接收器来提取卡夫卡的信息吗？另外，我能知道如何在Intellij IDEA中运行kafka星火流项目吗？
浏览 1提问于2014-09-10得票数 1
回答已采纳
2回答
我是Apache Nifi的新手，但我在尝试连接kafka微服务(与生产者)和Apache nifi消费者时遇到了一些问题。 基本上，我有一个这样的docker-compose： zookeeper:
container_name: zookeeper_test
image: wurstmeister/zookeeper #zookeeper:3.5.7
ports:
- 2181:2181
kafka:
container_name: kafka_test
image: wurstmeister/kafka #:2.13-2.6.0
ports:
- 90
浏览 37提问于2021-07-14得票数 0
我有一个非常简单的发送信息给Kafka的设置：
var producerConfig = new ProducerConfig
{
BootstrapServers = "www.example.com",
SecurityProtocol = SecurityProtocol.SaslSsl,
SaslMechanism = SaslMechanism.ScramSha512,
SaslUsername = _options.SaslUsername
浏览 10提问于2022-03-16得票数 0
1回答
我如何确保Pub/Sub模式中的公平性，例如kafka，当一个出版商生成数千条消息，而所有其他生产者的信息数字都很低的时候？谁的生产活动会很高，这是不可预测的。
如果其他制作人的消息不需要等待几个小时，仅仅因为一位制作人非常活跃，那就太好了。
它的模式是什么？卡夫卡或者谷歌PubSub这样的技术有可能吗？如果是，怎么做？
在这种情况下，多个分区也不能很好地工作，或者我可以看到它是如何工作的。
浏览 3提问于2021-06-10得票数 0
回答已采纳
1回答
我的maven POM文件创建了带有测试服务的docker容器。它还运行必须与这些容器中的服务对话的测试。容器可以使用docker-compose.yml文件中定义为链接的主机名相互交谈，例如：
kafka:
image: spotify/kafka
hostname: kafka
environment:
ADVERTISED_HOST: kafka
ADVERTISED_PORT: 9092
ports:
- "9092:9092"
our_service:
build: ../../logstash/
ports:
浏览 2提问于2017-10-26得票数 0
我有一个由2个Kafka代理组成的集群，以及一个具有复制因子2的主题。如果其中一个代理死亡，我的制作人是否能够继续向这个退化的一个节点集群发送新消息？或者复制因子2需要2个活动节点，而消息将被拒绝？
浏览 8提问于2016-08-17得票数 3
回答已采纳
是否可以在application.yml中仅为应用程序生成的主题启用spring.cloud.stream.kafka.streams.binder.auto-create-topics，而不是自动创建应用程序使用的主题？我希望我的制作人只负责创建主题。我正在使用spring-cloud-stream和kafka-streams绑定器。 在更一般的意义上，这是创建kafka主题的正确方法吗？
浏览 30提问于2020-11-24得票数 0
回答已采纳
1回答
我是卡夫卡的新手。我写了我的第一个消息制作人像这样
private Properties kafkaProps = new Properties();
kafkaProps.put("bootstrap.servers", "broker1:9092,broker2:9092");
kafkaProps.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
kafkaProps.put("value.serializ
浏览 2提问于2016-05-13得票数 1
回答已采纳
假设我的制作人正在将消息写到主题A...once中，消息在主题A中，我想将相同的消息复制到主题B中，这在kafka中可能吗？
浏览 2提问于2020-01-23得票数 2
我试图在nodejs中使用Kafka节点创建kafka制作人的消息块，但没有得到任何解决方案。因此，现在我正在使用Java创建一个Kafka流生成器，我需要发送大小大于1MB的大型消息。我怎样才能在卡夫卡制作者中创造出大量的信息，并消耗同样的信息呢？
浏览 0提问于2019-09-12得票数 0
我有一个Kakfa代理正在运行，我正在使用JMX监视它。
此代理是一个码头容器，运行于kafka-server-start.sh启动的进程，JMX端口公开了9999，并将其用作环境变量。
当我连接到JMX端口并试图列出所有域时，我会得到以下内容；
kafka
kafka.cluster
kafka.controller
kafka.coordinator.group
kafka.coordinator.transaction
kafka.log
kafka.network
kafka.server
kafka.utils
我看不到kafka.producer，这是可以理解的，因为这个卡夫卡经纪
浏览 0提问于2018-12-31得票数 2
回答已采纳
我刚开始接触卡夫卡。我已经通过了。它只为kafka流DSL提供数据/主题管理。有人能为kafka流的处理器API共享任何相同类型的数据管理链接吗？我对处理器API的用户和内部主题管理特别感兴趣。
TopologyBuilder builder = new TopologyBuilder();
// add the source processor node that takes Kafka topic "source-topic" as input
builder.addSource("Source", "source-topic")
在流处理
浏览 5提问于2017-10-26得票数 0
回答已采纳
1回答
我是卡夫卡的新手。当我试着动手的时候，如果有人能建议我如何理解这一点，我将面临两个problems.It会有很大的帮助。
当我从Eclipse运行Kafka时，当我运行我的生产者代码时，我能够通过运行控制台的“Kafka -控制台-消费. my”中的消费脚本来查看我发送给kafka代理的内容。但是，当我从Eclipse运行Consumer时，我无法在Eclipse控制台上看到任何内容。下面是我在Eclipse中运行的消费者代码。在我的生产者代码中，我刚刚发送了一个值(一个字符串)到与消费脚本中显示的相同主题。当我在消费脚本上执行“”时，我无法看到相同的值。
public class Simpl
浏览 1提问于2017-04-21得票数 0
我有一个非常简单的kafka制作人，它使用以下方法发送数据：
ProducerRecord producerRecord = new ProducerRecord<String, Integer>(topic, symbol, value);
producer.send(producerRecord);
基本上是一个键/值对，其中的值是一个整数。
如何在ksql中为此定义流？我试过：
create stream mystream (symbol varchar) with (value_format='integer', kafka_topic='myTop
浏览 1提问于2020-05-02得票数 1
回答已采纳
当自动主题创建时，我发现num.partitions的值是整个kafka集群上的分区号。
我在我的机器上启动了2个代理，brokerA和brokerB都有它自己的server.properties (都是num.partitons = 2)，然后我启动了一个制作人来给kafka写一条消息。我发现每个代理都创建了一个分区。
问题是，当集群中的代理都有自己不同的num.partitions时，一个主题的多少分区的逻辑将变得模糊。例如，brokerA num.partiton=2，brokerB num.partitions=1。当一个主题被自动创建时，这个主题将有多少个分区？
我已经测试过这个案子
浏览 7提问于2019-11-30得票数 0
回答已采纳
1回答
如果记录是由制作人单独发布/发送到Kafka主题的，那么每个记录/消息是否都有一个单独的offset#分配？还是将多个记录合并成将为其分配offset#的单个消息？
浏览 0提问于2018-10-10得票数 0
回答已采纳
这是一个卡夫卡批处理过程。我想读一个本地CSV文件，并将它写入一个Kafka主题。
然后消费者必须从他们订阅的主题中获取数据。
预期：，我希望将消耗的数据附加到格式的Parquet格式的文件中。请帮助我以有效的方式实现这一目标。
卡夫卡制作人投入：
卡夫卡消费者产出：
我希望将该值附加到HDFS中的文件中。
浏览 3提问于2020-03-05得票数 0
我有以下情况:我正在建设一个网站，其中有关于电影制作人和他们的电影的信息。我有一个节点类型的电影制作人，具有传记内容等，另一个为电影，与一个领域的“作者”，引用一个或多个电影制作者(因为一部电影可能是由他们中的几个)。
我还有一个名为“电影记录”的视图块，它列出了所有作者是用户正在观看的电影制作者(节点)的电影。用上下文过滤器设置这个非常简单。
但现在，我想在这个街区，以及电影的名称，所有的电影制作者可能已经制作了电影(在电影节点的‘作者’字段)，是不同的电影制片人被观看。显示由作者字段引用的所有电影制作者节点是即时的，但我想删除我在上下文过滤器中使用的电影制作人。我们的目标是得到这样的东西：
浏览 2提问于2015-04-29得票数 1
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
实时音视频
云直播活动推荐
腾讯云开发者
