JDBC Beam을 사용하여 Oracle에서 가져 오기

Question

아래 프로그램은 Oracle 11g에 연결하여 레코드를 가져 오는 것입니다. 어떻게 pipeline.apply()에서 코더에 대해 NullPointerException을 주는지.

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) 

Answer 1

안녕!

죄송합니다. 오류 메시지는별로 도움이되지 않지만 실제로는 유효성 검사 단계입니다. 이 문제를 개선하기 위해 BEAM-959을 제출했습니다.

당신은 우리가 빔에서 대부분의 다른 장소에서 가지고있는 것처럼,이 코더의 자동화를 개선하기 위해 BEAM-960을 제기 한

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

를 통해 같은 코더를 제공해야합니다.

Answer 2

시도해보십시오.

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)); 
       } 
       }) 
      )) 

그리고 Project에 mySql Connector jar를 추가하는 것을 잊지 마십시오. 미리 감사드립니다.