问题标签 [streamsets]

我正在尝试使用 JDBC 将数据从 Kafka 消费者写入 MySQL。我能够使用 JDBC 从 MySQL、PostgreSQL 等数据库中获取数据，但无法将数据写入数据库。但是我能够将数据写入文本文件。

我错过了什么或应该做什么？

我正在尝试使用 SQL 更改跟踪源来创建数据摄取管道。我已经连接了源并指定了所有必需的 jdbc 参数，并且管道验证成功。但是在运行管道时，我在日志中得到以下异常

仔细观察，我发现发送到 SQL Server 的查询是错误的

反对

问题是参数没有被引用。请任何人都可以指导我我做错了什么或如何减轻这种行为

在 cdc 的情况下，StreamSets Data Collector 有没有办法根据源数据库在目标数据库中自动创建表？

我正在从源读取数据：mssql 并写入目标 postgresql。如果我对源中的 50 个表感兴趣，我不想在目标数据库中手动创建这些表。

亲爱的，

我正在考虑如何在给定的通用数据中心架构中正确使用 Streamsets 的选项：

• 我有几种数据类型（csv、tsv、json、来自 IOT 的二进制文件）需要由 CDC 捕获并以原样格式保存到 Kafka 主题中，然后原样沉入 HDFS 数据湖。
• 然后，另一个 Streamsets Pipeline 将从这个 Kafka 主题中消费，并将其转换为通用格式（取决于数据类型）为 JSON，并执行验证、屏蔽、元数据等并保存到另一个 Kafka 主题。
• 相同的 JSON 消息将以 Avro 格式保存到 HDFS Data Lake 中以进行批处理。
• 然后，我将使用 Spark Streaming 使用相同的 JSON 消息进行实时处理，假设 JSON 数据已准备就绪，并且可以进一步丰富其他数据以进行可扩展的复杂转换。

我没有使用 Streamsets 进行进一步处理，也没有使用 Spark Streaming 进行可扩展的复杂转换，这不是 SLA 管理的一部分（因为 Spark 作业不是从 Streamsets 中触发的）此外，我无法在此设计中将 Kafka Registry 与 Avro 一起使用验证 JSON 模式和 JSON 模式是基于作为 Javascript 嵌入到 StreamSets 中的自定义逻辑进行验证的。

上面的设计有什么可以做得更好的？

提前致谢...

我刚刚升级了我的 MapR 集群，我正在尝试启动 Streamsets。但是我收到以下错误：

56 BGDTEST5 流集[24543]：在 com.streamsets.pipeline.BootstrapMain.main(BootstrapMain.java:291) 6 月 12 日 09:48:56 BGDTEST5 systemd[1]：sdc.service：主进程退出，代码=退出，状态=1/FAILURE Jun 12 09:48:56 BGDTEST5 systemd[1]: Unit sdc.service 进入失败状态。6 月 12 日 09:48:56 BGDTEST5 systemd[1]：sdc.service 失败。

我可以在streamsets-libs 目录中看到有问题的stage-lib。这以前可以无缝地工作。我究竟做错了什么？

我目前正在编写通过 CLI 在本地安装流集扩展的代码。我要编写的一项检查是确保扩展适用于本地安装的流集版本。

当我尝试从 CLI 查询版本时，这就是我遇到的问题。

除了遍历文件系统并查找文件之外，我如何确定安装了哪个版本的流集VERSION？

我需要使用 StreamSets 将 Phoenix 表中的记录转换为 JSON 文件。出于初始 POC 目的，我正在尝试从 Phoenix 进行简单的提取到文件中。源是指向 Phoenix 的 JDBC 查询消费者，目前，它直接连接到本地 FS 目标。在实际实现中，当然会有更多的组件，比如 Jython Evaluator。

它验证得很好。但是当我尝试运行它时，我收到以下错误：

管道状态：RUNNING_ERROR：com.streamsets.pipeline.api.StageException：JDBC_77 - ColumnNotFoundException 试图执行查询'SELECT <我选择的列> FROM <模式>。<表>'。根据阶段配置在 1 个错误后放弃。第一个错误：SQLState: 42703 错误代码：504 消息：错误 504 (42703)：未定义的列。列名=0

我基本上没有任何使用 StreamSets 或 Phoenix 的经验（是的，我想在这里稍微改进一下）。出现此错误是因为我将 JDBC 对象直接连接到本地 FS 吗？在 Informatica 中，我们必须为 Source 对象创建列模板。我在 JDBC 对象中看不到任何选项。除了以下字段外，我几乎将所有内容都保留为默认值：

• JDBC 连接字符串：jdbc:phoenix:ip-< some_ip >-south-1.compute.internal:2181/hbase-unsecure,ip-< some_ip >.ap-south-1.compute.internal:2181/hbase-unsecure, ip-< some_ip >.ap-south-1.compute.internal:2181/hbase-unsecure
• SQL 查询：SELECT <我选择的列> FROM <模式>.<表>
• 使用凭证 和增量模式：未选中
• 初始偏移和偏移列：0
• 根字段类型：列表映射

它变得如此令人沮丧。请帮忙！

我有一个 MSSQL 数据库，其结构通过 Postgres 数据库复制。我在 MSSQL 中启用了 CDC，并在 StreamSets Data Collector 中使用了 SQL Server CDC 客户端来监听该数据库表中的更改。

但我找不到在 Postgres 中写入相同表的方法。

例如，我在 MSSQL 中有 3 个表：tableA、tableB、tableC。我在 Postgres 中有相同的表。我将数据插入到 tableA 和 tableC 中。我希望通过 Postgres 复制这些更改。

在 StreamSets DC 中，为了写入 Postgres，我使用 JDBC Producer 并在我指定的表名称字段中：${record:attributes('jdbc.tables')}。

执行此操作，将从 tableA_CT、tableB_CT、tableC_CT 中读取数据。启用 CDC 选项时由 MSSQL 创建的表。所以我将在 ${record:attribute('jdbc.tables')} 中得到这些表名。

有没有办法在与 MSSQL 相同的表中写入 Postgres？

在我当前的项目中，我正在使用 StreamSets，我想使用 Hashicorp Vault 作为我的凭据存储，但是我无法credential:get()在任何我想使用的地方使用函数。例如，在 Azure IoT Hub Producer 块中的共享访问密钥中。我知道我可以使用运行时属性，但我认为它不能解决我的问题。我是否遗漏了某些内容，或者我credential:get()只能在标有钥匙图标的字段中使用？

我正在尝试在 StreamSets 中设置 Mysql-BinaryLog，但它抱怨它无法加载驱动程序实例。

我的.cnf：

为了使 StreamSets 起源工作，还需要做些什么吗？