用于处理数据的实时管道，插入到 PSQL答案

【问题标题】：Real-time pipeline for processing data, insert into PSQL用于处理数据的实时管道，插入到 PSQL
【发布时间】：2022-04-25 01:42:15
【问题描述】：

我有每天都会进来的文件，我想在它们进来时对其进行处理并插入到现有的 sql 表中（使用 postgres）。创建自动化管道的最佳方法是什么？

我已经在 python 上编写了文件处理脚本，它以格式返回要附加到 sql 表的数据。使此管道实时的最佳方法是什么。也就是说，让管道自动处理发送给我的文件，然后将数据添加到 sql 表中。目前我正在批量手动执行此操作，但我想完全自动化该过程。缺少的关键步骤是让脚本自动处理文件。我正在阅读 Apache kafka 可以提供帮助，但我在这里仍然是新手。

非常感谢任何帮助！

【问题讨论】：

如果您处理文件只是为了将它们加载到 postgres 中，我认为添加 Kafka 会引入不必要的复杂性。如果可以说服正在写入文件的系统改为写入 Kafka，那么它会更有意义。但是对于文件-> DB，请保持简单。恕我直言。
感谢您在此提供的意见 - 您会怎么做？我仍然希望这是实时的，即每当文件发送给我时，它都应该被自动处理。谢谢
数据从何而来？我会专注于尝试使那部分实时化；使用平面文件确实完全违背了实时性。使用 Kafka Connect，有数百个连接器可以将数据从外部系统拉入 Kafka。
感谢您的回复。数据由客户发送。它们是结构化文件 ASC X12（索赔文件）。我希望发送的每个文件都被自动处理并添加到数据库中（向发件人返回一条关于它成功的消息）。谢谢！

标签： python sql postgresql apache-kafka

【解决方案1】：

或者，您可以使用 JDBC 接收器连接器。它将数据从kafka实时传递到postgresql。可以通过 upsert 实现幂等写入。还支持自动创建表和有限的自动进化。 good connector description ; config-options

其实上面的connector会订阅一个kafka topic，并在stream中向数据库表发送kafka消息。您还可以订阅多个主题。

最低配置示例

{
  "name": "name_connector",
  "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
  "value.converter": "io.confluent.connect.avro.AvroConverter",
  "topics": [
    "topic_name"
  ],
  "connection.url": "postgresql://log:pass@hostname:5432/db",
  "dialect.name": "PostgreSqlDatabaseDialect",
  "insert.mode": "INSERT",
  "batch.size": "10",
  "table.types": [
    "table"
  ],
  "auto.create": true
}

【讨论】：