使用 Apache Beam 加载文件到数据库答案

【问题标题】：File to DB load using Apache beam使用 Apache Beam 加载文件到数据库
【发布时间】：2021-06-10 10:29:41
【问题描述】：

我需要将一个文件加载到我的数据库中，但在此之前我必须根据一些文件数据验证数据库中是否存在数据。例如，假设我在一个文件中有 5 条记录，那么我必须在数据库中检查 5 次以获取单独的记录。

那么我怎样才能动态获得这个值呢？我们必须在行中传递动态值而不是 2 (preparedStatement.setString(1, "2");)

在这里，我们正在创建一个 Dataflow 管道，它使用 Apache Beam 将数据加载到数据库中。现在我们创建一个管道对象并创建一个管道。使用 PCollection 我们存储到数据库中。

Pipeline p = Pipeline.create(options);
p.apply("Reading Text", TextIO.read().from(options.getInputFile()))
    .apply(ParDo.of(new FilterHeaderFn(csvHeader)))
    .apply(ParDo.of(new GetRatePlanID()))
    .apply("Format Result", MapElements.into(
        TypeDescriptors.strings()).via(
        (KV < String, Integer > ABC) - >
        ABC.getKey() + "," + ABC.getValue()))
    .apply("Write File", TextIO.write()
        .to(options.getOutputFile())
        .withoutSharding());

// Retrieving data from database
PCollection < String > data =
    p.apply(JdbcIO. < String > read()
        .withDataSourceConfiguration(JdbcIO.DataSourceConfiguration.create(
                "com.mysql.cj.jdbc.Driver", "jdbc:mysql://localhost:3306/XYZ")
            .withUsername("root")
            .withPassword("root1234"))

        .withQuery("select * from xyz where z = ?")
        .withCoder(StringUtf8Coder.of())
        .withStatementPreparator(new JdbcIO.StatementPreparator() {
            private static final long serialVersionUID = 1 L;
            @Override
            public void setParameters(PreparedStatement preparedStatement) throws Exception {
                preparedStatement.setString(1, "2");

            }
        })
        .withRowMapper(new JdbcIO.RowMapper < String > () {
            private static final long serialVersionUID = 1 L;
            public String mapRow(ResultSet resultSet) throws Exception {
                return "Symbol: " + resultSet.getInt(1) + "\nPrice: " + resultSet.getString(2) +

                    "\nCompany: " + resultSet.getInt(3);
            }
        }));

【问题讨论】：

如果您将数据库中的完整文件加载到临时表中，然后执行查询以仅合并您想要的字段，这不应该更有效吗？然后删除临时表。
我们是否有任何替代选项。因为我们有限制在生产中创建表。我们如何将其管理为代码？我们有相同的代码sn-p吗？

标签： google-cloud-platform apache-beam apache-beam-io apache-beam-internals

【解决方案1】：

按照建议，最有效的方法可能是将整个文件加载到临时表中，然后进行查询以更新必要的行。

如果无法做到这一点，您可以改为将表读入 Dataflow（即"select * from xyz"），然后执行 join/CoGroupByKey 以将记录与文件中找到的记录匹配。如果您希望现有数据库与您希望上传到其中的文件相比非常大，您可以有一个 DoFn 直接使用 JDBC 对您的数据库进行查询（可能在 DoFn 的 setUp 方法中缓存连接）而不是使用 JdbcIO。

【讨论】：