使用 Beam 和 DataFlow 将数据从 Cloud SQL 移动到 Elastic Search答案

【问题标题】：Moving data from Cloud SQL to Elastic Search using Beam and DataFlow使用 Beam 和 DataFlow 将数据从 Cloud SQL 移动到 Elastic Search
【发布时间】：2020-02-11 20:01:23
【问题描述】：

我是梁和谷歌数据流的新手，我创建了一个简单的类，通过在下面编写这个类，使用批处理将数据从云 sql 迁移到 Elastic Search：

package com.abc;

class DataFlowTest{

    public static void main(String[] args) {

    DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class);
        options.setProject("staging");  options.setTempLocation("gs://csv_to_sql_staging/temp");    options.setStagingLocation("gs://csv_to_sql_staging/staging");  options.setRunner(DataflowRunner.class);    options.setGcpTempLocation("gs://csv_to_sql_staging/temp");
            Pipeline p = Pipeline.create(options);


      p.begin();

      p.apply(JdbcIO.read().withQuery("select * from user_table").withDataSourceConfiguration(JdbcIO.DataSourceConfiguration.create("com.mysql.jdbc.Driver", "jdbc:mysql://"+EnvironmentVariable.getDatabaseIp()+"/" + EnvironmentVariable.getDatabaseName()+ "&socketFactory=com.google.cloud.sql.mysql.SocketFactory&user="+Credentials.getDatabaseUsername()+"&password="+Credentials.getDatabasePassword()+"&useSSL=false")));


  Write w = ElasticsearchIO.write().withConnectionConfiguration(
            ElasticsearchIO.ConnectionConfiguration.create(new String [] {"host"}, "user-temp", "String").withUsername("elastic").withPassword("password")
            );
   p.apply(w);

p.run().waitUntilFinish(); } }

and find below my dependnecies in pom.xml


<dependency>
      <groupId>org.apache.beam</groupId>
      <artifactId>beam-sdks-java-core</artifactId>
      <version>2.19.0</version>
    </dependency>

    <dependency>
      <groupId>org.apache.beam</groupId>
      <artifactId>beam-runners-google-cloud-dataflow-java</artifactId>
      <version>2.19.0</version>

      <exclusions>
      <exclusion>
      <groupId>com.google.api-client</groupId>
      <artifactId>google-api-client</artifactId>
      </exclusion>

      <exclusion>
      <groupId>com.google.http-client</groupId>
      <artifactId>google-http-client</artifactId>
      </exclusion>

      <exclusion>
      <groupId>com.google.http-client</groupId>
      <artifactId>google-http-client-jackson2</artifactId>
      </exclusion>


      </exclusions>

    </dependency>

    <dependency>
    <groupId>org.apache.beam</groupId>
    <artifactId>beam-sdks-java-io-elasticsearch</artifactId>
    <version>2.19.0</version>
    </dependency>

    <dependency>
        <groupId>org.apache.beam</groupId>
        <artifactId>beam-sdks-java-io-jdbc</artifactId>
        <version>2.19.0</version>
    </dependency>



    <dependency>
      <groupId>org.apache.beam</groupId>
      <artifactId>beam-sdks-java-io-google-cloud-platform</artifactId>
      <version>2.19.0</version>
      <exclusions>
      <exclusion>
      <groupId>io.grpc</groupId>
      <artifactId>grpc-core</artifactId>
      </exclusion>

      </exclusions>

    </dependency>

现在问题是它说的编译错误：

Pipeline 类型中的方法 apply(PTransform) 不适用于参数 (ElasticsearchIO.Write) 在那一行： p.apply(w);

有人可以帮忙吗？我在 pom 文件中做了一些排除来修复一些依赖冲突

【问题讨论】：

您是否尝试过调用PTransform，将Write w = ElasticsearchIO.write()...p.apply(w) 替换为p.apply(ElasticsearchIO.write()...)？
是的，结果还是一样
我正在尝试了解您的用例，如果我的理解有误，请告诉我。因此，您正在尝试从 cloudSql 读取数据并将结果数据写入 Elastic Search。对？如果是这种情况，那么我可以为您提供帮助。
是的，这正是我想要做的，我已经通过了这个错误，我现在面临另一个问题；已经在这里发布stackoverflow.com/questions/60197434/…
@Tamer Saleh，如果你已经解决了当前的问题，那么写下答案是一个好习惯，以帮助下一个贡献者研究类似问题的解决方案。

标签： google-cloud-dataflow google-cloud-sql apache-beam

【解决方案1】：

无法直接将 ElasticSearchIO.write 应用于管道对象。首先创建一个 PCollection，然后将 ElasticsearchIO 应用到 PCollection。请参考以下代码。

PCollection<String> sqlResult1 = p.apply(
                JdbcIO.<String>read().withDataSourceConfiguration(config).withQuery("select * from test_table")
                        .withCoder(StringUtf8Coder.of()).withRowMapper(new JdbcIO.RowMapper<String>() {

                            private static final long serialVersionUID = 1L;

                            public String mapRow(ResultSet resultSet) throws Exception {
                                StringBuilder val = new StringBuilder();
                                return val.append(resultSet.getString(0)).append(resultSet.getString(1)).toString();
                                // return KV.of(resultSet.getString(1), resultSet.getString(2));
                            }
                        }));

        sqlResult1.apply(ElasticsearchIO.write().withConnectionConfiguration(ElasticsearchIO.ConnectionConfiguration
                .create(new String[] { "https://host:9243" }, "user-temp", "String").withUsername("").withPassword("")));

我认为上面的代码应该适用于您的用例。

【讨论】：