recordField的Kafka连接接收器分区，单位为Ticks
，是指在Kafka消息队列中，对于recordField字段的数据进行分区的时间间隔单位。
Kafka是一种分布式流处理平台，用于发布和订阅消息的持久性事件流。在Kafka中，分区是用于水平扩展和增加吞吐量的机制。每个分区都可以在不同的节点上进行复制，以提供高可用性和容错性。
recordField是指在消息中包含的特定字段，通常是与业务相关的数据。Kafka连接接收器负责从数据源中接收和处理消息，并将其写入Kafka消息队列。分区是指将消息分配到Kafka主题的不同分区中，以便实现负载均衡和并行处理。
单位Ticks是Kafka中的时间单位，代表了系统的时钟周期。Ticks的长度可以根据系统的配置进行调整，通常以毫秒为单位。将分区的时间间隔单位设置为Ticks意味着分区的时间间隔将与系统的时钟周期相对应。
根据具体的应用场景和需求，可以根据数据的产生速率和处理能力来设置分区的时间间隔。较短的时间间隔可以提供更精细的消息分区和更高的并发处理能力，但也会增加系统的负载和资源消耗。较长的时间间隔则可以减少系统的负载和资源消耗，但可能会导致消息处理的延迟和吞吐量下降。
对于recordField的Kafka连接接收器分区，单位为Ticks，推荐使用腾讯云的云原生消息队列 CMQ（Cloud Message Queue）服务。CMQ提供可靠的、高可用的消息队列服务，适用于分布式消息驱动的应用场景。CMQ支持根据消息的属性进行消息分区，并提供了丰富的管理和监控功能。更多关于腾讯云CMQ的信息可以查看腾讯云官方网站：https://cloud.tencent.com/product/cmq
相关·内容
有人能解释一下partition.duration.ms和flushsize在以下配置中的意义吗？设置这些属性的背后应该是什么想法？
"connector.class": "io.confluent.connect.s3.S3SinkConnector",
"s3.region": "eu-central-1",
"partition.duration.ms": "1000",
"topics.dir": "root_bucket",
"flu
浏览 0提问于2018-10-11得票数 4
回答已采纳
我有一个连接器，它使用两个主题。
一个主题有6个分区，第二个主题有2个分区(总共有8个分区)
当我将tasks.max设置在6下时，要使用的分区在任务之间很好地平衡(通过查看consumer_group)
但是，如果我在示例8中将tasks.max设置为高于6，那么在consumer_group中有两个没有主题属性的任务(所有8个任务都处于运行状态)，因此有2个空闲任务
主题offset.storage.topic只有6个分区
连接器不可能拥有比主题offset.storage.topic的分区数更多的活动(不仅是正在运行)任务？
那么，值offset.storage.
浏览 4提问于2021-05-11得票数 0
回答已采纳
卡夫卡连接节点之间是否共享分区？我的意思是，两个节点可以写到同一个分区吗？问这个问题，因为在我们的例子中，几乎所有的记录都以相同的键进入kafka，并且由于密钥是相同的，所有的记录都存储在同一个分区中。虽然我们有两个连接节点，但其中只有一个连接到目标DB。是因为只允许一个节点写入分区吗？如果没有，如何使两个节点写入同一个分区？
浏览 10提问于2022-07-25得票数 0
回答已采纳
我遇到了一些问题，编写从接收器子任务到输出kafka主题的关键流。
作业的形式是:来源->过滤器-> keyby(id) ->平面映射->接收器
这些异常来自kafka生产者并导致检查点超时：
FlinkKafkaException: Failed to send data to Kafka: Failed to allocate memory within the configured max blocking timeFlinkKafkaException: Failed to send data to Kafka: Expiring 16 record(s) fo
浏览 2提问于2019-10-10得票数 1
1回答
我们希望使用Kafka连接接收器连接器，以便将消息从Kafka复制到Mongo。在我们的用例中，我们有多个主题，每个分区都有一个分区(主题的名称可以重新定义，例如topic.XXX.name)。这些主题的数量在不断增加。我想知道Kafka连接架构是否适合这个用例。如果是这样的话，如何配置它的高可伸缩性和并行性？tasks.max是什么？工人人数？
浏览 7提问于2022-06-10得票数 1
我试图使用Kafka Sink连接器将批处理的数据发送到NOSQL数据库。我正在跟踪文档，并对在何处实现发送记录的逻辑感到困惑。请帮助我理解如何在内部处理这些记录，以及在批处理中必须使用什么Put()或Flush()来处理记录。
浏览 5提问于2017-07-02得票数 8
回答已采纳
我在一个3节点的纱线集群上运行了一个星火流(createStream API)应用程序，其中每个节点都有128 each (!)该应用程序读取卡夫卡主题的记录并写入HDFS。
大多数情况下，应用程序由于Java堆错误而失败/被杀死(大部分是接收方失败)，不管我为executor/驱动程序配置了多少内存。
16/11/23 13:00:20 WARN ReceiverTracker: Error reported by receiver for stream 0: Error handling message; exiting - java.lang.OutOfMemoryError: Java
浏览 2提问于2016-11-23得票数 0
回答已采纳
2回答
是否可以用spark-streaming-kafka-0-10库配置Spark来读取多个Kafka分区，或者用单个任务来读取整个Kafka主题，而不是为每个可用的Kafka分区创建不同的Spark任务？
请原谅我对这些技术的粗略理解；我认为我对火花和卡夫卡还不熟悉。架构和设置主要是为了探索和查看这些技术是如何协同工作的。
我有四个虚拟主机，一个有火花放电的主人，每个都有一个火花工作者。其中一个主机还运行一个基于的Kafka代理。每台主机都有四个核心和大约8GB的未使用RAM。
Kafka broker有206个主题，每个主题有10个分区。因此，总共有2,060个分区可供应用程序读取。
我正在使
浏览 3提问于2017-02-20得票数 1
回答已采纳
我已经部署了kafka connect。
Connect分布式正在运行合流3.1.2。(在4台设备中作为一台group.id运行。)
我已经给了tasks.max=4，但是我不知道为什么我只有一个任务。
浏览 1提问于2018-05-03得票数 0
回答已采纳
我有一个卡夫卡连接水槽。在我的主题中，我有用刻度表示的字段，而不是正确的时间戳。我最终希望将其用作目标中的分区字段(在本例中为Azure数据湖gen 2)。 我尝试过将TimeBasedPartitioner与timestamp.extractor和时间戳转换器一起使用，但它只是在格式上出错。在我看来-所有这些时间戳转换器都使用“时间戳”字段，而我的是以刻度为单位的，所以在我可以使用时间戳转换器之前，我必须做额外的转换，但我不确定为什么，作为我调查过的SMT，没有提供任何这样的东西。 我得到的错误 java.lang.IllegalArgumentException: Invalid for
浏览 18提问于2021-04-20得票数 0
回答已采纳
我们已经在一个项目上使用了一段时间的Kafka，目前只使用。我很难理解“任务”在Kafka中的作用，特别是这个连接器。我理解“连接器”；它们包含一组关于特定源/接收器的配置，以及它们连接到的主题。我理解连接器和任务之间存在着1:Many关系，以及任务用于并行化工作的一般原则。但是，我们如何理解连接器何时会/可能创建多个任务？
在源连接器的情况下，我们使用JDBC连接器通过时间戳和/或主键来获取源数据，因此这在本质上似乎是连续的。事实上，我们所有的源连接器似乎只有一个任务。什么会触发卡夫卡连接来创建多个连接器？目前，我们正在中运行Kafka，但是只有一个工作人员；如果我们有多个工作人员，那么
浏览 3提问于2021-04-26得票数 6
回答已采纳
我希望生产和部署我的Kafka Connect应用程序。然而，我有两个关于tasks.max设置的问题，这是必需的，而且非常重要，但是对于实际设置这个值的细节是模糊的。
如果我有一个具有n个分区的主题，我希望从该主题中使用数据并将其写入某个接收器(在我的示例中，我正在写入S3)，我应该将tasks.max设置为什么？我应该将它设置为n吗？我应该将其设置为2n吗？直觉上，我似乎想要将值设置为n，这就是我一直在做的事情。
如果我改变我的Kafka主题并增加主题上的分区怎么办？我将不得不暂停我的卡夫卡连接器，并增加tasks.max，如果我将它设置为n？如果我将值设置为2n，那么我的连接器应该自动增
浏览 4提问于2017-01-28得票数 28
回答已采纳
我们在消费者内部存在性能问题。更准确地说，是指对消息做了什么。
通常，我们得到2000行的文件，拆分为6000个Kafka事件。
我们的主题X配置了10个分区。在这种情况下，我们可以在一个给定组中启动最多10个使用者，每个使用者将附加到一个单独的分区，直到重新平衡。顺便提一下，在我们的例子中，分区中的消息排序是强制的。
问题是，每条消息最多需要6秒才能被消费者处理。所以整个文件大约在1小时内被处理完。
我还知道Kafka是为可伸缩性而设计的，这种可伸缩性是由主题的分区数量来管理的。
(一个显而易见的解决方案是重构消费者的代码。这是有计划的，但我们开发团队的工作负载是这样的，而且这种重构可能在几
浏览 1提问于2019-03-06得票数 0
我有一个关于将数据下沉到Kafka主题中的流作业和另一个消耗kafka主题中的数据的流作业。我的Kafka集群的分区号配置为3。 当我将作业的并行度设置为4时，只有3个插槽在忙于生成数据，并且只有3个消费者子任务获得了数据。 考虑到任务槽数量的限制，我想将并行性改为1。但是，当我将作业的并行度设置为1时，只有1 consumer task slot获得了数据。 在我看来，即使我将并行度设置为1，我仍然将数据下沉到3个分区中，并且可以有3个消费子任务使用数据。 为什么只有一个消费子任务在这里工作？
浏览 230提问于2021-06-16得票数 0
回答已采纳
如果我是正确的，默认情况下，星火流1.6.1使用单个线程从每个Kafka分区读取数据，假设我的Kafka主题分区为50，这意味着每个50个分区中的消息将按顺序读取或可能以循环方式读取。
案例1：
-If是的，那么如何在分区级别并行读取操作？创建多个KafkaUtils.createDirectStream是唯一的解决方案吗？
e.g.
val stream1 = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](
ssc, kafkaParams, topicsSe
浏览 4提问于2016-12-12得票数 2
我需要为卡夫卡连接创建一个定制的分区器。我使用kotlin扩展了自定义类中的：
class RawDumpHourlyPartitioner<T> : HourlyPartitioner<T>() {
...
}
并相应地更改连接器配置以使用自定义类：
"partitioner.class": "co.myapp.RawDumpHourlyPartitioner",
然后，我创建了jar (我们使用)，并根据kafka连接映像将其包含到定制的坞映像中(图像版本与我们在项目中使用的依赖项相同)：
FROM gradle:6.0-jdk8 a
浏览 7提问于2020-02-13得票数 2
回答已采纳
我的火花流版本是2.0，kafka版本是0.10.0.1，火花流-kafka-0-10_2.11。我使用直接的方式获得卡夫卡的记录，我现在想限制我在一批中获得的信息的最大数量。因此，我设置了max.poll.records值，但它不起作用。星星之火中的消费者数量是卡夫卡中的分区数?那么火花流中记录的最大数量是max.poll.records*consumers？
浏览 1提问于2018-09-27得票数 2
如果我的卡夫卡主题收到这样的记录
CHANNEL | VIEWERS | .....
ABC | 100 | .....
CBS | 200 | .....
我有Spark结构化的流媒体代码来读取和处理Kafka记录如下：
val spark = SparkSession
.builder
.appName("TestPartition")
.master("local[*]")
.getOrCreate()
import spark.implicits._
浏览 0提问于2018-04-20得票数 5
当我试图读取从Kafka到S3的消息时，我在向Kafka连接类路径添加jar时遇到了问题。
目标是基于时间戳在分区中写入消息，时间戳是Kafka消息中键的一部分。
为了简短起见，我必须提供自定义的时间戳提取器。在文档之后，创建了一个实现TimestampExtractor接口的类，并将一个JAR位置添加到plugin.path属性中。
问题是，当我启动连接时，就找不到类。不知怎的，罐子不在门廊里，我
org.apache.kafka.common.config.ConfigException: Invalid timestamp extractor: partitioner.SpotadDat
浏览 1提问于2017-12-18得票数 4
回答已采纳
2回答
我已经启动了我的zookeeper和Kafka服务器。我启动了我的Kafka制作器，它发送了10条主题为'xxx‘的消息。然后叫住了我的卡夫卡制作人。现在我启动了我的Kafka用户，并订阅了主题'xxx‘。我的消费者使用我的Kafka生成器发送的10条消息，该生成器现在没有运行。我需要我的Kafka消费者应该只消费运行Kafka服务器的消息。有什么方法可以做到这一点吗？以下是我的消费者属性中的内容。
props.put("bootstrap.servers", "localhost:9092");
String consumeGroup
浏览 1提问于2016-09-30得票数 3
我有一个冲槽代理(1个源，1个通道，1个接收器)，从特定的目录(Spooldir)读取数据并发送到Kafka Sink。根据我的观察，它只是写入特定的分区。
启用并行发送/处理到kafka主题/分区的配置是什么？
浏览 1提问于2017-05-02得票数 0
我正在对Kafka进行调查，以评估它是否适合我们的用例。你能帮我理解一下Kafka在改变现有集群大小时有多灵活吗？
我正在对Kafka进行调查，以评估它是否适合我们的用例。你能帮我理解一下Kafka在不拆掉集群的情况下向现有集群添加brokers的灵活性吗？这样做的时候有什么需要注意的吗？
浏览 2提问于2019-01-08得票数 0
您好，我正在尝试使用pyspark和kafka创建一个neo4j接收器，但由于某些原因，此接收器在neo4j中创建了重复项，我不确定为什么会发生这种情况。我希望只得到一个节点，但看起来它正在创建4个节点。如果有人有想法，请让我知道。
Kafka生产者代码：
from kafka import KafkaProducer
import json
producer = KafkaProducer(bootstrap_servers='10.0.0.38:9092')
message = {
'test_1': 'test_1',
&
浏览 0提问于2020-03-20得票数 1
我计划动态缩放一个Flink应用程序。该应用程序使用kafka-flink连接器消耗来自Kafka的事件。
因为应用程序的“热身”只需几分钟(缓存.)而改变并行级别涉及重新启动，我更喜欢提交(放大)或交替杀死(缩小)任务，而不是改变并行级别。
我想知道，从性能、逻辑和执行计划来看，这种方法和Flink内置的并行执行有什么不同吗？
换句话说，10个相同的Flink任务与一个并行级别= 10的任务( env.setParallelism( 10 ) )有什么区别？
浏览 1提问于2019-09-25得票数 0
回答已采纳
我使用S3 Sink连接器将卡夫卡的记录写入S3。最终，我将使用Kafka从我的数据库中捕获CDC数据包，然后将这些数据包写入S3。
但是，我不希望每个CDC数据包(在Kafka中是一个记录)都写入到一个单独的S3对象中。我希望配置一个基于大小或基于时间的条件，以便每隔X秒或Y字节就将所有记录写入一个S3对象。
我还没有找到任何东西可以将记录写到一个对象，但是我发现fetch.min.bytes和fetch.max.wait.ms每隔X秒或Y字节就写一次对象，但是多条记录仍然被写成单独的对象。
浏览 4提问于2022-11-14得票数 0
1回答
我正在尝试使用与星火结构的流查询。我得到的错误是，用户在处理数据时找不到合适的偏移量。如果没有此触发器，查询将正常运行(如预期的那样)。
我所做的：
阅读卡夫卡主题的数据：
val inputStream: DataStreamReader = spark
.readStream
.format("kafka")
.option("kafka.bootstrap.servers", "host:port")
.option("subscribe", "input_topic&
浏览 6提问于2021-10-06得票数 2
1回答
Spring中的套接字连接数
、、、、
我试着理解当使用春季卡夫卡连接到Kafka时，套接字连接是如何打开的。下面的代码连接到一个名为test的Kafka主题，该主题有32个分区。
application.yml
spring:
kafka:
bootstrap-servers: testbroker:9092
consumer:
enable-auto-commit: false
auto-offset-reset: earliest
spring.kafka.consumer.key-deserializer: org.apache.kafka.
浏览 4提问于2022-03-27得票数 0
我正在使用带有Kafka的Spark结构化流，主题被订阅为模式：
option("subscribePattern", "topic.*")
// Subscribe to a pattern
val df = spark
.readStream
.format("kafka")
.option("kafka.bootstrap.servers", "host1:port1,host2:port2")
.option("subscribePattern", "topic.*&
浏览 0提问于2020-12-18得票数 2
回答已采纳
VoltDB的Kafka连接器支持读取avro消息吗?下面是我配置连接器的尝试吗？
{
"name":"KafkaSinkConnector",
"config":{
"connector.class":"org.voltdb.connect.kafka.KafkaSinkConnector",
"tasks.max":"1",
"voltdb.servers":"node1:21212,node2:21212,
浏览 1提问于2018-10-02得票数 0
1回答
我试图使用FlinkKafkaProducer010生成元素，但是当我打开消费控制台窗口时，元素似乎出现了故障。
我创建的主题使用: kafka-topics.bat --创建--主题mytopic --动物园管理员localhost:2181 -分区1-复制因子1
使用者是使用:kafka-控制台- The .The-动物园管理员localhost:2181 -主题mytopic创建的
我使用的Kafka生产者代码是：
public static void main(String[] args) throws Exception {
ParameterTool parameterTo
浏览 1提问于2018-05-28得票数 0
在连接器post请求创建过程中，我收到以下错误：
Error message : 500 Internal Server Error,
error response: {"error_code":500,"message":"org.apache.kafka.connect.runtime.distributed.RebalanceNeededException: Request cannot be completed because a rebalance is expected"}
Requst body:
{"name"
浏览 22提问于2019-07-16得票数 0
1回答
我知道Kafka JDBC Sink Connector对于数组数据类型有一些缺点。但是，可以将Sink连接器与支持数组数据类型的简单Kafka连接器结合使用。如何从Kafka配置中过滤出来，切换成简单的Kafka Connector配置？Kafka Connect如何支持数组字段
name: topic_name
type: array
item: Topic file
是否可以将其作为字符串而不是数组使用到db
"fields":[{
"name":"item_id",
"type":{
"type&
浏览 0提问于2021-02-25得票数 2
1回答
我有一个关于Kafka-Connect的一般性问题。我翻阅了文档和博客，但找不到直接的答案。 如果有两个工作者，运行单个连接器(实例)，那么连接器(实例)如何决定何时产生新任务，例如。tasks.max = 10？另外，连接器(实例)如何决定要派生多少任务，如果。tasks.max = 10？它依赖于底层硬件配置吗？例如：核心或内存或cpu利用率的数量？
浏览 16提问于2020-07-08得票数 0
我在python中一起玩星火流和kafka，并且松散地跟随着，但是我对前面提到的KafkaUtils.createStream()函数有点困惑。
通过显式解释主题字典的影响，并没有做太多的工作。但我怀疑，我之所以这么想，只是因为我对卡夫卡作品的了解是不可靠的，答案是显而易见的。
我理解它应该是这样的字典：{"topic.name": 1}和我可以调出文档，这意味着创建的流将从单个分区中消耗。
所以我想我只是想澄清一下这个特殊功能的用法，以及我对卡夫卡概念的理解。我们将使用以下示例：
假设我已经定义了一个主题my.topic，它有3个分区，传入的消息在一个键上被拆分，让我们简单地
浏览 1提问于2018-01-09得票数 3
回答已采纳
我正在使用结构化流从Kafka读取数据，我需要将数据保存到InfluxDB。在基于Dstreams的常规方法中，我这样做了如下：
val messages:DStream[(String, String)] = kafkaStream.map(record =>
(record.topic, record.value))
messages.foreachRDD { rdd =>
rdd.foreachPartition { partitionOfRecords =>
val influxService = new InfluxService()
浏览 0提问于2018-05-17得票数 2
我正在介绍kafka，我想知道如何像这样使用从topic中的分区到ClickHouse表的消息：
当我有3个主题时，很容易将每个主题的表连接起来
ENGINE = Kafka SETTINGS
kafka_broker_list = 'broker:9092',
kafka_topic_list = 'topic1',
kafka_group_name = 'kafka_group',
kafka_format = 'JSONEachRow'
但是我不知道如何将消息从一个主题的分区消费到表中。请帮帮忙
浏览 1提问于2019-01-15得票数 0
我在用debezium捕获mysql中的数据时遇到了问题，将数据捕获并使用kafka连接jdbc接收器将其消费到另一个mysql。
因为debezium对kafka主题产生的模式和有效负载与kafka连接jdbc接收器所期望的模式不兼容。
当jdbc接收器想要使用数据并在另一个mysql中创建记录时，我就会异常。
我该如何解决这个问题？
浏览 2提问于2017-05-14得票数 1
回答已采纳
我正在尝试使用s创建一个卡夫卡接收器连接器。但是，出于某种原因，日志输出正在报告一个SourceConnectorConfig：
INFO ConnectorConfig values:
connector.class = com.spredfast.kafka.connect.s3.sink.S3SinkConnector
key.converter = null
name = transactions-s3-sink
tasks.max = 1
transforms = null
value.c
浏览 0提问于2017-10-21得票数 0
回答已采纳
我试图使用s3接收器卡夫卡连接器将一些json数据存储到s3中。我的json格式如下：
{
"server": someserver,
"id": someid,
"time": "2018-01-18T23:47:03.737487Z"
}
我想根据数据所处的时间来划分数据，但忽略分钟和秒。例如：以上json将属于2018-01-18T23目录。我应该如何在属性文件中设置field.partition以实现这一点？
非常感谢!
浏览 1提问于2018-01-24得票数 0
回答已采纳
"log.retention.bytes“是我们用来保留主题消息日志的参数，我给出的值为1073741824。
我参考了卡夫卡文档，它说"log.retention.bytes“中给出的大小是每个分区，所以这意味着假设我使用的所有主题都有20个分区，那么根据文档，卡夫卡将保留的字节的总大小是20*1073741824。
但我需要的是清晰
Will Kafka retain 20*1073741824 bytes for all the topics?
(or)
Will Kafka retain 20*1073741824 bytes
浏览 1提问于2019-03-27得票数 0
1回答
强迫汇流s3水槽冲洗
、、、
我设置了kafka连接s3接收器，持续时间设置为1小时，还设置了一个相当大的冲洗计数，比如10,000次。现在，如果卡夫卡通道中没有多少消息，s3接收器将尝试在内存中缓冲它们，然后等待它们累积到刷新计数，然后将它们一起上传，并将偏移量提交给自己的使用者组。
但想想这种情况。如果在频道里，我只发送5,000条消息。然后没有s3水槽冲洗。然后，经过很长一段时间，这5,000条信息最终将被驱逐出卡夫卡，因为保留时间。但是这些消息仍然存在于s3接收器的内存中，而不是在s3中。这是非常危险的，例如，如果我们重新启动s3接收器或运行s3接收器的机器只是崩溃。然后我们失去了那5000条信息。我们无法再从卡夫
浏览 3提问于2018-06-08得票数 4
回答已采纳
在kafka连接中，所有的接收器连接器都将使用不同的组，并对connect-connector_name进行命名转换。但是我想使用一个自定义名称作为前缀。(我们可以在接收器配置- name properties中这样做，但默认情况下需要设置它)
我试图在consumer.properties文件中设置它，但是没有用。
有人知道它是怎么设置的吗？另外，如果我为我的所有接收器连接器设置了一个组，会发生什么？
浏览 0提问于2021-08-26得票数 0
回答已采纳
我想使用Kafka将数据从本地流式传输到云(S3)。为此，我需要在源机和云上安装kafka。但我不想把它放在云端。我需要一些S3连接器，通过它我可以与kafka连接，并将数据从本地传输到云。
浏览 1提问于2017-03-29得票数 2
我正在考虑使用Kafka连接器和创建我自己的Kafka消费者/生产者来将一些数据从Kafka转移到Kafka，我看到了Kafka连接器在可扩展性和容错方面所提供的价值。但是，如果“任务”由于某种原因而失败，我无法找到连接器的确切行为。以下是几种场景：
对于接收器连接器(S3-Sink)，如果它(任务)成功地将数据发送到目标(例如由于网络问题)失败(在所有重试之后)，那么工作人员会发生什么情况？它会坠毁吗？它能够再消费同样的数据从Kafak以后吗？
对于源连接器(JDBC )，如果它不能发送给Kafka，它以后会重新处理相同的数据吗？这取决于来源是什么？
以上问题的答案是否取决于
浏览 0提问于2018-08-22得票数 1
在kafka连接分布式模式下，需要通过REST提交kafka连接源和kafka连接接收器配置。卡夫卡如何利用卡夫卡集团协调人组建消费集团？
浏览 5提问于2022-09-19得票数 0
在Spark2.2中，我使用了使用Kafka的星火流，如下所示：
val conf = new SparkConf()
.setAppName("Test")
.setMaster("local[*]")
val sc = new SparkContext(conf)
val ssc = new StreamingContext(sc, Seconds(60))
val kafkaParams = Map[String, String](
"metadata.broker.list&
浏览 5提问于2017-12-13得票数 2
回答已采纳
我目前正在spark上运行一个流媒体应用程序，它使用新的直接方法(没有接收器)消费一个简单的kafka主题。由于我们在Kafka集群上遇到了一些问题，即使在Kafka上恢复正常后，流媒体应用程序也遇到了一些问题:一些作业正在处理几乎空的rdd，而另一些作业正在处理两倍大小的rdd。平均输入速率仍然与以前相同，但处理时间因此不规律。
我试着重启应用程序，但没有什么不同。我在Kafka和Spark两边都没有发现任何堆栈或线索。
有人能告诉我如何解决这类问题吗？或者至少到哪里去看一看，以了解发生了什么？
浏览 0提问于2016-05-18得票数 0
我正在读书。
它说：
作为接收器，upsert连接器可以使用更改流。它将将INSERT/UPDATE_ with数据写入普通Kafka消息值，并将DELETE数据写入为带空值的Kafka消息(为键指示墓碑)。
它没有提到，如果UPDATE_BEFORE消息是为了重新插入卡夫卡，那么会发生什么呢？
在同一链接()中，文档提供了一个完整的示例：
INSERT INTO pageviews_per_region
SELECT
user_region,
COUNT(*),
COUNT(DISTINCT user_id)
FROM pageviews
GROUP BY user_regio
浏览 0提问于2021-06-09得票数 0
回答已采纳
我不明白为什么我只能在spark UI页面(8080)中看到一个KafkaReceiver，但是在Kafka中我确实有10个分区，我在spark集群中使用了10个内核，我的代码在python中也是这样: kvs = KafkaUtils.createStream(ssc，zkQuorum，ssc，ssc，“火花流-消费者”，{zkQuorum: 10}) --我认为KafkaReceivers数字应该是10而不是1。我真的很困惑。提前谢谢你！
浏览 1提问于2015-06-26得票数 4
回答已采纳
我需要将MySQL数据库复制到PostgreSQL数据库。我选择了：
Debezium连接
Avro格式
合流模式注册表
卡夫卡
数据正在被复制，但是，我正在丢失一些模式信息。例如，在Postgres中将具有mysql格式的datetime格式的列复制为bigint，没有创建外键，也没有保留列的顺序(这很好)，等等。
PostgreSQL接收器连接器：
{
"name": "jdbc-sink-dbt",
"config": {
"connector.class": "
浏览 1提问于2022-01-14得票数 0
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
实时音视频
即时通信 IM活动推荐
腾讯云开发者
