任何人都可以帮助我使用流式传输的连续数据中的数据。主题的雪花连接器中应该给出什么？
因为我能够通过具有所需主题名称的单个表数据填充数据。但是我需要捕获连续的数据流到表中。
【问题讨论】：
标签： snowflake-cloud-data-platform
为了将数据流式传输到 Snowflake，我们利用 Apache NiFi 从 Kafka 中提取、修改/转换、保存到 S3 存储桶，然后启动雪管（使用 NiFi 处理器）以摄取到 Snowflake 表。但是一旦在 Snowflake 中，我们利用 dbt 来进行更多的 sql 转换和扩充。
【讨论】：
来自文档：
Kafka 连接器完成以下流程来订阅 Kafka 主题并创建 Snowflake 对象：
Kafka 连接器根据通过 Kafka 配置文件或命令行（或 Confluent 控制中心；仅限 Confluent）提供的配置信息订阅一个或多个 Kafka 主题。
连接器为每个主题创建以下对象：
一个用于临时存储每个主题的数据文件的内部阶段。
一个管道用于提取每个主题分区的数据文件。
每个主题一个表格。如果为每个主题指定的表不存在，则连接器创建它；否则，连接器会在现有表中创建 RECORD_CONTENT 和 RECORD_METADATA 列，并验证其他列是否可以为空（如果不是，则生成错误）。
然后摄取过程如下：
一个或多个应用程序将 JSON 或 Avro 记录发布到 Kafka 集群。记录被分成一个或多个主题分区。
Kafka 连接器缓冲来自 Kafka 主题的消息。当达到阈值（时间或内存或消息数量）时，连接器会将消息写入内部阶段的临时文件。连接器触发 Snowpipe 以摄取临时文件。 Snowpipe 将指向数据文件的指针复制到队列中。
Snowflake 提供的虚拟仓库通过为 Kafka 主题分区创建的管道将暂存文件中的数据加载到目标表（即主题配置文件中指定的表）中。
连接器监控 Snowpipe 并在确认文件数据已加载到表中后删除内部阶段中的每个文件。如果失败阻止数据加载，连接器会将文件移动到表阶段并生成错误消息。
连接器重复步骤 2-4。
https://docs.snowflake.com/en/user-guide/kafka-connector-overview.html
【讨论】：