JDBC Fetch из оракула с Beam

Приведенная ниже программа предназначена для подключения к Oracle 11g и извлечения записей. Как бы то ни было, это дает мне исключение NullPointerException для кодера в pipe.apply().

Я добавил ojdbc14.jar в зависимости проекта.

public static void main(String[] args) {

        Pipeline p = Pipeline.create(PipelineOptionsFactory.create());      
         p.apply(JdbcIO.<KV<Integer, String>>read()
                   .withDataSourceConfiguration(JdbcIO.DataSourceConfiguration.create(
                          "oracle.jdbc.driver.OracleDriver", "jdbc:oracle:thin:@hostdnsname:port/servicename")
                   .withUsername("uname")
                   .withPassword("pwd"))
                   .withQuery("select EMPID,NAME from EMPLOYEE1")
                   .withRowMapper(new JdbcIO.RowMapper<KV<Integer, String>>() {
                     public KV<Integer, String> mapRow(ResultSet resultSet) throws Exception {
                       return KV.of(resultSet.getInt(1), resultSet.getString(2));
                     }
                   }));
         p.run();

    }

дает следующую ошибку. Любая подсказка?

Exception in thread "main" java.lang.NullPointerException: coder
    at com.google.common.base.Preconditions.checkNotNull(Preconditions.java:228)
    at org.apache.beam.sdk.io.jdbc.JdbcIO$Read.validate(JdbcIO.java:283)
    at org.apache.beam.sdk.io.jdbc.JdbcIO$Read.validate(JdbcIO.java:216)
    at org.apache.beam.sdk.Pipeline.applyInternal(Pipeline.java:399)
    at org.apache.beam.sdk.Pipeline.applyTransform(Pipeline.java:307)
    at org.apache.beam.sdk.values.PBegin.apply(PBegin.java:47)
    at org.apache.beam.sdk.Pipeline.apply(Pipeline.java:158)
    at org.apache.beam.examples.v030.JdbcUtil.main(JdbcUtil.java:21)

apache-beam apache-beam-io
person naga    schedule 10.11.2016    source источник

Ответы (2)


arrow_upward
3
arrow_downward

Всем привет!

Извините, сообщение об ошибке не очень полезно, но на самом деле это шаг проверки. Я подал BEAM-959, чтобы улучшить это.

Вы должны предоставить кодировщик, например через

.withCoder(KvCoder.of(VarIntCoder.of(), StringUtf8Coder.of())`

Я подал BEAM-960, чтобы улучшить автоматизацию этого кодировщика, как мы в большинстве других мест в Beam.

person Kenn Knowles    schedule 10.11.2016
comment
Я отредактировал ваше сообщение выше после внесения изменений выше. Это дает следующую ошибку. Исключение в потоке main driver.T4CConnection.isValid(I)Z - person naga; 11.11.2016
comment
Есть ли что-то, что мне не хватает? - person naga; 14.11.2016
comment
@ Kenn, чтобы решить проблему, дайте мне знать, если вам нужна дополнительная информация. - person naga; 15.11.2016
comment
Ваша дальнейшая проблема не связана с этим вопросом. Это может быть связано с подготовкой или версией драйвера JDBC. Можете ли вы открыть новый вопрос для этого? (и, возможно, принять этот ответ :-) - person Kenn Knowles; 15.11.2016

arrow_upward
-1
arrow_downward

Попробуй это.

 pipeline.apply(( JdbcIO.<KV<Integer, String>>read().withCoder(KvCoder.of(VarIntCoder.of(),StringUtf8Coder.of())) 
               .withDataSourceConfiguration(JdbcIO.DataSourceConfiguration.create(
                      "com.mysql.jdbc.Driver", "jdbc:mysql://localhost:3306/deepakgoyal")
                    .withUsername("root")
                    .withPassword("root"))
               .withQuery("select empid, name from employee")

               .withRowMapper(new JdbcIO.RowMapper<KV<Integer, String>>() {
                 public KV<Integer, String> mapRow(ResultSet resultSet) throws Exception {
                   return KV.of(resultSet.getInt(1), resultSet.getString(2));
                 }
               })
             ))

И не забудьте добавить jar коннектора MySQL в свой проект.

person Dpk Goyal    schedule 18.01.2017
comment
Извините, но я не понял, что именно вы хотите. - person Dpk Goyal; 18.01.2017