Kafka数据流ID
是指Apache Kafka中用于唯一标识数据流的标识符。Kafka是一种高吞吐量、低延迟的分布式消息队列系统，被广泛应用于大规模数据处理和实时流处理场景。
Kafka数据流ID的分类：
- Topic ID：Kafka中的数据流被组织成一个个主题（Topic），每个主题都有一个唯一的标识符，即Topic ID。Topic ID用于在Kafka集群中标识和定位特定的数据流。
Kafka数据流ID的优势：
- 高吞吐量：Kafka通过分布式架构和批量处理机制，能够实现每秒处理数百万条消息的高吞吐量。
- 低延迟：Kafka的设计目标之一是提供低延迟的消息传递，使得实时数据处理成为可能。
- 可靠性：Kafka采用分布式副本机制，确保数据的可靠性和容错性，即使在节点故障的情况下也能保证数据不丢失。
- 可扩展性：Kafka的分布式架构支持水平扩展，可以根据需求增加节点和分区，以应对不断增长的数据流量。
Kafka数据流ID的应用场景：
- 实时数据处理：Kafka作为一个高性能的消息队列系统，广泛应用于实时数据处理场景，如日志收集、实时监控、实时分析等。
- 大数据集成：Kafka可以作为数据源和数据目的地，实现不同数据系统之间的数据集成和数据传输，如将数据从数据库导入到Hadoop集群。
- 流式处理：Kafka与流处理框架（如Apache Flink、Apache Spark Streaming）结合使用，可以实现实时流处理，对数据进行实时计算和分析。
- 事件驱动架构：Kafka的发布-订阅模型使得它成为构建事件驱动架构的理想选择，可以实现解耦和异步通信。
推荐的腾讯云相关产品和产品介绍链接地址： 腾讯云提供了一系列与消息队列相关的产品和服务，可以满足不同场景下的需求，以下是其中几个推荐的产品：
- 云消息队列 CMQ：腾讯云的消息队列服务，提供高可靠、高可用的消息传递能力，支持多种消息模式和消息协议。产品介绍链接：https://cloud.tencent.com/product/cmq
- 云原生消息队列 TDMQ：腾讯云的云原生消息队列服务，基于Apache Pulsar开源项目构建，提供高性能、低延迟的消息传递能力。产品介绍链接：https://cloud.tencent.com/product/tdmq
- 数据流引擎 CDE：腾讯云的数据流引擎服务，提供实时数据处理和流式计算能力，支持Kafka作为数据源和数据目的地。产品介绍链接：https://cloud.tencent.com/product/cde
请注意，以上推荐的产品和链接仅供参考，具体选择应根据实际需求和情况进行评估和决策。
相关·内容
1回答
我已经开始学习MQTT，因为我在目前的组织中有一个远程信息处理的用例。我想将MQTT代理(蚊子)消息集成到我的kafka中。
由于每个工具都在单个组织内的MQTT代理中以其自己的主题发送数据，因此我希望将所有这些数据推送到kafka中。现在我知道用kafka创建这么多主题(超过一百万)是不可取的。此外，我不想将所有车辆数据保存在一个卡夫卡主题中，因为我希望稍后将所有这些数据放在S3中，通过车辆id进行区分。
我如何才能做到这一点，而不是在kafka中创建如此多的主题。一种方法是kafka的消费者将事件分离并放入S3中，但我相信在s3中会有很多小文件。
浏览 19提问于2019-11-28得票数 1
我已经开发了一个Flink程序，可以读取Twitter上的推文，并将它们推送到Kafka上。然后，它从Kafka获取tweet并对其进行处理。
"Tweets“转换从tweet的文本中提取标签和用户，并在默认输出中发出它们，并在副输出中发出它们的每一对。
附加的图像是从Flink Web UI中拾取的。我不明白为什么Kafka源和Tweets处理操作符合并到一个任务中，主要是我希望Tweets接收器接收所有来自Kafka源的原始Tweet，而不是tweets处理操作符的输出。
程序是正确的吗？
这是代码的相关部分：
FlinkKafkaConsumer010<Strin
浏览 1提问于2018-06-30得票数 1
我无法在http | log之后部署和执行基本的流
wget https://raw.githubusercontent.com/spring-cloud/spring-cloud-dataflow/2.1.0.RELEASE/spring-cloud-dataflow-server/docker-compose.yml;
export DATAFLOW_VERSION=2.1.0.RELEASE;
export SKIPPER_VERSION=2.0.2.RELEASE;
docker-compose up;
open http://localhost:9393/dashboard/#/st
浏览 2提问于2019-05-24得票数 5
回答已采纳
从服务器上，我能够连接并从配置了SSL的远程kafka服务器主题中获取数据。 在GCP中，如何使用Google Dataflow管道通过SSL信任库、密钥库证书位置和Google服务帐户json连接到远程kafka服务器？ 我使用Eclipse plugin for dataflow runner选项。 如果我指向GCS上的证书，当证书指向Google存储桶时，它会抛出错误。 Exception in thread "main" org.apache.beam.sdk.Pipeline$PipelineExecutionException: org.apache.kafka.
浏览 13提问于2019-01-24得票数 1
回答已采纳
我正在尝试并行读取Kafka消息，从而并行处理它们。我的Kafka主题有10个分区。我正在尝试创建5个DStreams，并应用Union方法对单个DStream进行操作。以下是我到目前为止尝试过的代码：
def main(args: scala.Array[String]): Unit = {
val properties = readProperties()
val streamConf = new SparkConf().setMaster("local[2]").setAppName("KafkaStream")
val s
浏览 0提问于2017-09-16得票数 0
1回答
我正在与阿帕奇光束一起工作。我的任务是从kafka主题中提取数据，并在数据流中进行处理。数据流是否支持kafkaIO？
KafkaIO支持哪些runners？
浏览 0提问于2020-07-08得票数 2
我正在尝试配置分布式Kafka connect，但在Azure HD Insight中找不到任何用于配置Kafka connect的jars。
你能帮我解决上面的问题吗？
浏览 14提问于2021-06-01得票数 0
1回答
我正在设计卡夫卡的实时消费工作，并考虑春季启动批处理和弹簧引导。
kafka版本: 2.11-1.1.1jdk : 1.8
哪种更好些呢？
弹簧引导批+ kafka 弹簧引导+kafka
请告诉我原因:)
浏览 3提问于2019-10-11得票数 1
我正在构建一个Apache管道，作为一个无限制的源从Kafka读取。
我能够使用直接跑步器在本地运行。
但是，当在云中使用运行时，附加的异常堆栈跟踪将导致管道失败。
似乎最终是抛出了。我不太清楚该如何解决这个问题。
java.io.IOException: Failed to start reading from source: org.apache.beam.sdk.io.kafka.KafkaUnboundedSource@33b5ff70
com.google.cloud.dataflow.worker.WorkerCustomSources$UnboundedReade
浏览 0提问于2018-11-05得票数 5
回答已采纳
为什么你用卡夫卡来谈论数据流。
然而，例如，在REST API方面，我从未听说过数据流这样的东西。
也许有人能告诉我术语数据流在Kafka中的真正含义。顺便说一句，我知道在Kafka中有生产者和消费者。生产者将数据发送到代理，然后代理将其发送给消费者。
浏览 0提问于2021-05-02得票数 1
1回答
我有1000个rest web服务。需求将数据摄取到数据湖的标准方法？满载、CDC和流数据。将数据加载到kafka主题中，不重复
浏览 0提问于2018-01-30得票数 1
我有一个需求，我需要在Azure事件中心使用kafka主题。需要创建将使用作为参数提供的主题的POST端点。消息必须以kafka主题为属性，以消息内容为主体，在公共子主题上发送。
这是一个很高的要求。我看过以了解如何实现这一目标。但是，如果有人实时地实现了这一功能，即从Azure事件中心到google的事件，或者曾经进行过类似的实现，请提供帮助。
浏览 1提问于2021-06-03得票数 1
回答已采纳
给定来自Kafka的数据流，是否有内置的方法，或者执行SQL查找的最有效方法是什么。 例如，将流对象连接到sql表，并将其与Kafka数据流连接。给定可以在自组织基础上更新的sql表，
浏览 16提问于2020-04-15得票数 0
回答已采纳
3回答
我是Apache NiFi的初学者，但到目前为止，我读到的所有教程都是关于kafka与Nifi的集成。kafka如何成为Nifi的补充？为什么我们不直接使用Nifi来推送我们的消息，而不是使用kafka？
注意:我看过的所有教程都没有提到这一点。
浏览 7提问于2018-11-29得票数 14
回答已采纳
在纱线上的Spring XD :版本1.2.1直接绑定支持kafka源代码。
1.我知道这还不支持(从1.3.0版本开始)，任何确定的日期/版本将有助于我们的项目进度？
2.这种对kafka源代码支持的直接绑定对我们的项目非常关键。正因为如此，我们在我们的项目中完全放弃了Spring XD纱线。
试着去做
流创建--名称directkafkatohdfs --定义"kafka | hdfs“流部署directkafkatohdfs --属性"module.*.count=0”
命中异常“必须是正数，0-当前不支持kafka源计数”
我只想消除消息总线/传输(redis/kafk
浏览 0提问于2015-11-27得票数 0
2回答
我已经安装了Kafka connect，使用confluent-4.0.0，使用hdfs连接器，我可以将从Kafka topic接收的Avro记录保存到hive。我想知道是否有任何方法可以在写入hdfs接收器之前修改记录。我的要求是对记录的值做一些小的修改。例如，在整数上执行算术运算或操作字符串等。请建议是否有任何方法可以实现这一点
浏览 2提问于2018-03-07得票数 2
1回答
我目前正在使用VoltDB kafka importer从多个kafka主题导入数据。我正面临着装载机的性能问题。我阅读了VoltDB文档，但无法找到如何微调进口商。
如何指定主题的特定分区？
我的当前设置
6 nodes of VoltDB cluster and Kafka importers on the nodes with custom procedure for insert.
Kafka导入程序配置
Host: 172.x.x.x:9092
Topic: mytopic_1,mytopic_2,...mytopic_10
Procedure: tinsert
创建过程INSER
浏览 1提问于2018-09-07得票数 1
我正在尝试使用python code.My代码中的kafka.ReadFromKafka()方法读取kafka主题中的数据，如下所示：
from apache_beam.io.external import kafka
import apache_beam as beam
options = PipelineOptions()
with beam.Pipeline(options=options) as p:
plants = (
p
| 'read' >> kafka.ReadFromKafka({
浏览 1提问于2020-07-07得票数 0
我有一个类似于主题的问题。如何在Spring Boot中建立React JS和Apache Kafka之间的连接？ 我不知道如何从React JS前端连接Apache Kafka和订阅主题。 我能用这个吗？类似下面的示例： let socket = new WebSocket("ws://localhost:9092");
浏览 5提问于2020-03-26得票数 0
是否有任何开箱即用的解决方案来连接opc-da和Kafka？我希望从和opc-da服务器获得数据到apache Flink
浏览 3提问于2020-04-14得票数 3
到目前为止，我一直在使用Spring应用程序(使用Stream)和Kafka，而没有任何支持基础设施(PaaS)。
由于我们的公司平台运行在Kubernetes上，我们需要将那些Spring应用程序移动到K8s中，以允许应用程序进行扩展等等。显然，每个应用程序都有多个实例，因此我们将为每个应用程序定义一个使用者组，以确保每个消息的唯一传递和处理。
Kafka将在Kubernetes外面跑。。
现在我的疑问是:由于部署在k8s上的应用程序是通过抽象底层荚的k8s服务访问的，而且单个应用程序荚不能直接在k8s集群之外访问，所以卡夫卡将不知道如何调用使用者组的各个实例来传递消息，对吗？
我怎么能让他
浏览 2提问于2017-07-21得票数 1
回答已采纳
1回答
我有一个关于卡夫卡的的问题。我喜欢这个特性，因为在我的例子中，它意味着我可以使用Kafka作为无限的存储空间(例如，使用gcs后端)。但是，让我们假设由于某种原因，Kafka集群被删除，并且Kafka数据丢失。
gcs/s3存储中的数据仍然有用吗？
我的意思是，我可以将旧的日志插入到新的Kafka集群中，或者它现在完全没有用了吗( to级的日志)？
顺便说一句，我知道我可以分析gcs/S3存储中的段并提取数据。但这有点老生常谈，这就是为什么我想看看我是否能找到一个干净的解决方案。
浏览 3提问于2020-10-05得票数 1
我在春季云数据流中工作，在那里我有一个场景，比如从数据库中读取数据，并使用@In量InboundChannelAdapter将数据发送到kafka主题
下面是我遵循的策略。
如果列表为空，则使用->Created公共列表来存储对象
->if列表中有我不会投票的数据
->i将使用索引一个接一个地将值发送给卡夫卡，然后我将删除索引。
如果我保留@Bean，它只是将列表中的第一个对象插入到kafka主题中。
{"id":101443442，“名称”：“Mobile1”，“价格”：8000}
如果我删除@Bean，那么它将将所有空数据插入kafka中。
{}
浏览 4提问于2020-12-11得票数 0
在已经部署在Stream中的SpringBoot-application中，我使用了yaml-file中的属性：
spring:
kafka:
producer:
retries: 8
是否可以在Spring Cloud DataFlow的UI中显示和管理该属性，如下所示：
浏览 0提问于2020-12-29得票数 0
我想了解一下spring云数据流的体系结构。特别是使用它来编排纯粹的批处理管道。我已经在k8s中部署了所有组件，并且注意到我可以在没有中间件(Kafka / RabbitMQ)的情况下部署skipper组件，并通过服务器组件成功地运行批处理作业。
根据我可以看到的文档，skipper用于编排流作业。当纯粹用于批处理作业或任务时，skipper (以及随后的Kafka/RabbitMQ中间件)是否是春季云数据流的必要部分？
浏览 0提问于2020-04-23得票数 0
回答已采纳
我有一个IBM AIX的例子，需要将syslog、文件日志和可能的appslog发送到Apache，这有可能吗？我试着搜索一些关于nxlog的内容，但是他没有输出到Kafka的AIX，还有其他解决方案吗？
浏览 5提问于2018-01-05得票数 1
这个项目是为real time search engine - log analysis性能。
我有一个从Spark processing到Kafka的实时流媒体数据。
现在有了Kafka的输出，我想get the data from the Kafka using Flask..和visualize it using Chartjs或其他一些可视化工具..
如何从Kafka using the python flask获取实时流数据？
你知道我该怎么开始吗？
任何帮助都将不胜感激！
谢谢!
浏览 0提问于2016-02-05得票数 3
1回答
我试图使用从Kafka获取数据到Bigquery。我的数据流模板基于PythonSDK2.42+容器注册表+ apache_beam.io.kafka。
这是我的管道：
def run(
bq_dataset,
bq_table_name,
project,
pipeline_options
):
with Pipeline(options=pipeline_options) as pipeline:
kafka = pipeline | ReadFromKafka(
浏览 16提问于2022-11-06得票数 0
回答已采纳
对于使用KafkaAvroDeserializer的Kafka (消费者)，我需要在部署时使用--propertiesFile覆盖一些信任。KafkaBinder 1.1.1添加了这个配置选项。
如果可以的话，我不喜欢使用部署属性，因为它们在仪表板上是不可见的，而且如果另一个用户决定停止并重新启动流而不设置属性，这是很危险的。是否可以将这些属性包含在流定义中？我的多次尝试都失败了。我在参考文档上没有看到这样的例子。
例如相当于yaml的：
<...>
binder:
configuration:
value.deserializer: io.confluent.kafk
浏览 5提问于2017-01-31得票数 1
回答已采纳
4回答
我是nifi的新手，正在尝试理解(因为它看起来像是基于GUI的)是否有一种方法可以在Nifi上自动扩展和缩小，以及如何将xml Nifi模板部署到集群中。
本质上，我们试图做的是使用Nifi从kafka服务器收集JMX和日志文件，当它们以自动方式出现时，因此日志和JMX计数器开始流动到，比如说和Elasticsearch集群。
例如，现在我们已经使用Ansible自动化了kafka服务器的部署，所以我们希望在集群部署的同时配置和部署Nifi xml teamplates，这样就不需要手动干预来收集kafka jmx计数器和日志文件。
如何在Nifi中实现自动化？
浏览 0提问于2017-02-18得票数 5
我不确定我是否理解了Spring Cloud Data Flow (SCDF)和Apache Kafka Broker(及其相关的Zookeeper)之间的关系。
我阅读了一些文档，并观看了一个关于Spring Cloud数据流(SCDF)的网络研讨会。我了解不同的绑定选项(RabbitMQ，Kafka，...)用于此框架中不同微服务之间的通信。
我找到了很多关于如何使用Kafka的信息，但不知道在哪里运行/操作它。我的假设是否正确，即您必须独立于SCDF安装/管理/操作Kafka代理(+ Zookeeper)？SCDF“只是”使用Kafka代理进行内部通信，以及-可选的-用于微服务内的业务逻
浏览 0提问于2017-10-25得票数 1
1回答
当我们在基于某个键的流上应用group函数时，kafka如何计算这个值，因为在不同的分区中可能存在相同的键？我看到了()函数，它基本上对数据进行了重新划分，但我不明白这意味着什么。它会将所有具有相同键的消息移动到一个分区中吗？另外，我们可以通过()方法调用多久一次？如果有要求，我们能在收到每条消息后打电话给它吗？请建议一下。谢谢
浏览 1提问于2020-05-18得票数 1
回答已采纳
1回答
我试图找到Spring数据流和Azure服务总线设置的一些例子。。
我已经找到了，但它仍然在RC中，我没有看到任何示例(涵盖Spring数据流)。
你能帮我理解一下，如果我可以一起使用Spring数据流和Azure服务总线吗？
我能够使用Kafka和RabbitMQ运行示例，但我找不到任何可以用作Spring数据流集成解决方案的Azure服务总线
浏览 0提问于2019-07-05得票数 1
回答已采纳
1回答
我在活动上用“kafka -node”创建了kafka消费者。
consumer.on('message' ()=>{
connecting to mongodb and inserting to a collection.
})
用于创建与mongo的连接并返回对象的mongo.js文件
const MongoClient = require('mongodb').MongoClient, assert = require('assert');
const url = 'mongodb://root:****@ds031257.
浏览 0提问于2019-04-03得票数 0
回答已采纳
我正在开发一个火花应用程序，它需要从Kafka读取数据。我创造了一个卡夫卡主题，制片人在那里发布信息。我从控制台使用者那里验证消息已成功发布。
我写了一个简短的火花应用程序来读取Kafka的数据，但是它没有得到任何数据。以下是我使用的代码：
def main(args: Array[String]): Unit = {
val Array(zkQuorum, group, topics, numThreads) = args
val sparkConf = new SparkConf().setAppName("SparkConsumer").setMaster(&
浏览 2提问于2016-11-28得票数 1
2回答
我正在尝试使用同步拉API来使用Google PubSub消息。这可以在IO连接器库中获得。我想使用KafkaIO将使用的消息写到Kafka。我想使用FlinkRunner来执行作业，因为我们在GCP之外运行这个应用程序。
我面临的问题是，在GCP PubSub中，所消耗的消息没有得到ACK。我已经确认本地Kafka实例具有从GCP PubSub中使用的消息。GCP DataFlow中的文档表明，当管道被一个数据接收器终止时，数据包就会完成，在我的例子中，这就是Kafka。
但是，由于代码在Apache中运行，而不是在GCP DataFlow中运行，所以我认为某些类型的回调与ACK‘in提交的
浏览 3提问于2020-11-08得票数 0
回答已采纳
在不使用汇流平台的情况下，可以使用kafka连接将主题数据插入到db。我们有许可证问题。无论我在哪里做RnD，都找不到没有汇合的。请帮帮我！
浏览 0提问于2019-07-15得票数 0
回答已采纳
我使用的是Flink v1.4.0。我使用如下代码所示的Kafka FLink Consumer来使用Kafka主题中的数据：
Properties properties = new Properties();
properties.setProperty("bootstrap.servers", "localhost:9092");
// only required for Kafka 0.8
properties.setProperty("zookeeper.connect", "localhost:2181");
prope
浏览 6提问于2018-01-25得票数 3
回答已采纳
在使用ValueProvider执行run命令时，我尝试获取服务器ID作为参数 对于选项界面中的值提供程序： ValueProvider<String> getKafkaServer();
void setKafkaServer(ValueProvider<String> value); withBootstrapServers抛出错误“不兼容的类型:无法将org.apache.beam.sdk.options.ValueProvider转换为java.lang.String” PCollection<String> kafkaMessages =
浏览 47提问于2021-10-01得票数 0
我希望使用火花结构流从卡夫卡读取数据，并处理它，并写入蜂窝表。
val spark = SparkSession
.builder
.appName("Kafka Test")
.config("spark.sql.streaming.metricsEnabled", true)
.config("spark.streaming.backpressure.enabled", "true")
.enableHiveSupport()
.getOrCreate()
val events = sp
浏览 5提问于2018-02-06得票数 1
是否可以为Apache Kafka配置Spring Cloud DataFlow的身份验证？我可以在哪里看到示例？
谢谢。
浏览 15提问于2016-08-03得票数 0
我正在尝试完成第一个教程，但在本教程中我没有得到这个结果。
本教程让我将curl用于http源，并在日志接收器中看到带有stdout文件尾的结果。
我看不出结果。我在日志中看到了启动。
我尾随日志停靠程序exec -it跳尾-f /path/from/stdout/textbox/in/dashboard
我输入curl -H“内容-类型:文本/普通”-d“快乐流”
我只看到
2020-10-05 16:30:03.315 INFO 110 --- [ main] o.a.kafka.common.utils.AppInfoParser : Kafka ver
浏览 1提问于2020-10-05得票数 0
回答已采纳
如果应用程序使用Java将事件作为行插入到BigQuery表中，那么我希望避免意外地多次插入相同的事件/行(因为错误、重试等等)。
每个事件，我作为一行插入，有一个唯一的UUID，但我没有流偏移量。(它不能保证事件在我的应用程序中以固定的顺序到达，因为它使用的是带有多个分区的Kafka主题。)
使用，我能够使用以下函数的id参数依赖于。
com.google.cloud.bigquery.InsertAllRequest.RowToInsert.of(String id, Map<String, ?> content)
新的 (在提交模式下)确实有一个精确的一次保证，但前提是你可以
浏览 10提问于2022-02-06得票数 0
我有apache访问日志文件，我想在cassandra表中存储每个页面的访问计数(总数/每天/每小时)。
我正在尝试使用kafka连接从日志文件流到kafka主题。为了在Cassandra中增加指标计数器，我可以再次使用Kafka Connect吗？否则，这里应该使用哪些其他工具，例如kafka streams，spark，flink，kafka connect等？
浏览 0提问于2017-10-11得票数 0
我很难理解GCP、数据流/Apache和之间的区别。我想要做的是转向一种更适合于流数据处理的云本地解决方案，这样我们的开发人员就可以更多地关注核心逻辑的开发，而不是管理基础设施。
我们有一个现有的流解决方案，它由spring云数据流‘模块’组成，我们可以独立地迭代和部署，就像微服务一样，它工作得很好，但是我们希望迁移到GCP中的一个现有平台，该平台由我们的业务提供，需要我们使用GCP Dataflow。在高层次上，解决办法很简单：
流1:
Kafka Source (S0) -> Module A1 (Ingest) -> Module B1 (Map) -> Module
浏览 2提问于2019-11-26得票数 1
回答已采纳
1回答
有一个将值写入数据库的遗留服务。我需要将值转换为事件，然后将其发送到kafka。 我要做一个服务，一旦在固定的延迟中检查并发送新记录，还会将提交的记录is写入技术表，但也许还有其他方法、最佳实践或模式。 ?
浏览 20提问于2020-11-06得票数 1
回答已采纳
我遇到的问题是使用(SCDF)通过环境将数据移动到正确的位置。情况如下：
我有一个生产版的卡夫卡，我只能订阅(只读)。(3节点集群: node 1:9092，Node 2:9092，Node 3:9092)i有一个运行在Kubernetes集群中的SCDF实例，该集群使用RabbitMQ主干(全部在一台服务器上)i有一个节点卡夫卡服务器，用于另一家公司获取数据(ssl受保护)。
在查看SCDF时，我没有看到一种将Kafka定义为源并将其读入处理器(将是自定义的)并写入另一个Kafka实例的方法。
浏览 3提问于2021-06-03得票数 0
My管线: Kafka ->数据流流(Beamv2.3) -> BigQuery
考虑到低延迟在我的例子中并不重要，我使用来降低成本，如下所示：
BigQueryIO.writeTableRows()
.withJsonSchema(schema)
.withWriteDisposition(WriteDisposition.WRITE_APPEND)
.withCreateDisposition(CreateDisposition.CREATE_IF_NEEDED)
.withMethod(Method.FILE_LOADS)
.withTriggeringF
浏览 0提问于2018-03-19得票数 5
我对Flink相当陌生，在技术用例方面需要一些帮助。
当前场景：
我有一个在GKE上运行的flink应用程序，并使用自定义接收器将记录(从Kafka源)写入BigQuery。我可以将记录写入BigQuery，而不会出现任何问题。目前，记录被一个接一个地写入接收器中，因此每个kafka消息都会得到自己对BigQuery的插入api调用，这并不理想，因为我们需要执行批量插入，并且单独插入每个记录将非常昂贵。为此，我使用了bigquery-storagewrite。
新要求：
在将记录写入BigQuery之前，理想情况下，我希望在将它们写入接收器之前根据大小/时间对记录进行缓冲。
我不知道如何在Fl
浏览 5提问于2022-10-18得票数 2
回答已采纳
1回答
我是卡夫卡的新手，我用卡夫卡阅读了有关数据处理和进一步分析(例如火花)的东西，但对将数据放入卡夫卡的实际过程一无所知。我知道我得把它当成制片人。但是，例如，我如何从一个网络应用程序中发送跟踪数据，并将其应用到kafka中。我是否应该使用nginx日志文件作为能够直接将其写入kafka的生成器或服务器(例如，龙卷风和kafka的python )？如何创建一个非常简单的分析工具，从get请求中获取数据，并将其放入kafka进行进一步处理？
任何评论或评论，也只是一些小小的暗示，都会帮助我理解这件事。
浏览 12提问于2014-12-11得票数 0
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
