有没有办法用 Kafka 消费中的最新消息替换旧消息（避免最终 df 中重复）

Question

我正在使用来自某个主题的数据，正如我们所知，我们实时获取数据，其中我们看到重复的元素，如何实际上用最新消息替换旧消息。

我使用以下相同的代码从主题中使用

schema = StructType(
    [
        StructField("Id",StringType(),True),
        StructField("cTime",StringType(),True),
        StructField("latestTime",StringType(),False),
        StructField("service",StringType(),True),
    ]

topic = "topic1"
bootstrap_servers = "mrdc.it.com:9093,mrdc.it.com:9093,mrdc.it.com:9093"

options = {
    "kafka.sasl.jaas.config": 'org.apache.kafka.common.security.plain.PlainLoginModule required username="xxxxx.aud.com" password="xxxxxxxx";',\
    "kafka.ssl.ca.location": "/tmp/cert.crt",\
    "kafka.sasl.mechanism": "PLAIN",\
    "kafka.security.protocol" : "SASL_SSL",\
    "kafka.bootstrap.servers": bootstrap_servers,\
    "failOnDataLoss": "false",\
    "subscribe": topic,\
    "startingOffsets": "latest",\
    "enable.auto.commit": "false",\
    "auto.offset.reset": "false",\
    "enable.partition.eof": "true",\
    "key.deserializer": "org.apache.kafka.common.serialization.StringDeserializer",\
    "value.deserializer": "org.apache.kafka.common.serialization.StringDeserializer"
}
kafka_df = spark.readStream.format("kafka").options(**options).load()

kafka_mobile_apps_df = kafka_df.select(from_json(col("value").cast("string"), schema).alias("apps"))

df = avertack_kafka_eventhub_connections(source= "KAFKA", kafka_config=kafka_config)

sql_features = ["apps.Id",
                "apps.cTime",
                "apps.latesTime", 
                "apps.service"
               ]

kafka_df_features = df.selectExpr(sql_features)
display(kafka_df_features)

输出如图所示

Id              cTime                   latestTime              service
3178    2022-03-03T20:39:52.889Z    2022-03-03T20:39:58.601Z    mobile
3178    2022-03-03T20:39:52.889Z    2022-03-03T20:39:59.012Z    mobile
3240    2022-03-03T20:39:59.140Z    2022-03-03T20:39:59.220Z    mobile
3246    2022-03-03T20:40:00.615Z    2022-03-03T20:40:00.648Z    mobile
.
.
.

我们如何用第 2 行覆盖第 1 行，使用键作为 [“id”]，其中“latestTime”列，如何仅保留最新时间消息。

有没有实时的方法，如果没有，我们如何至少每小时检查一次，用新的

最终输出替换旧消息

Id              cTime                   latestTime              service
3178    2022-03-03T20:39:52.889Z    2022-03-03T20:39:59.012Z    mobile
3240    2022-03-03T20:39:59.140Z    2022-03-03T20:39:59.220Z    mobile
3246    2022-03-03T20:40:00.615Z    2022-03-03T20:40:00.648Z    mobile
.
.
.
.

Answer 1

Spark 更多地用于批处理/微批处理；它获取记录集合，您可以按时间排序并获取“最新”，或者您可以“按 id 分组”并执行相同操作...

但是，您必须将其与某些数据库/持久存储结合起来才能创建“最新，按 id”的视图。我见过使用 HBase、Couchbase、MongoDB 等完成此操作。如果需要的话，它们都具有一定程度的 Spark 集成。

开箱即用的 Spark，我认为它不容易提供此功能（您可以查看一个 RocksDB State Store）。

---

另外，Kafka 本身还有 Kafka Streams，它可以提供您正在寻找的内容，尽管它是使用 Java 完成的。

如果您必须使用Python，则可以设置KSQLdb，并且可以从任何其他语言使用其API，并允许您更轻松地将流处理定义为SQL语句。

---

Kafka Connect 是另一种选择，例如，如果您有一个标准的关系数据库，并且数据中的 ID 是表的主键，那么匹配的键将执行 UPDATE 查询并覆盖现有记录，或插入新的 ID。