如何创建从postgres到镶木地板的管道？

提问者：小点点

如何创建从postgres到镶木地板的管道？

我们正在创建一个数据流管道，我们将从postgres中读取数据并将其写入parque文件。我们正在使用org.apache.beam.sdk.io. jdbc读取和org.apache.beam.sdk.io.parque包来写入文件。ParquetIO.Sink允许您将GenericRecords的PCollection写入Parquet文件（从这里https://beam.apache.org/releases/javadoc/2.5.0/org/apache/beam/sdk/io/parquet/ParquetIO.html）。

这是我到目前为止的代码：

Pipeline p = Pipeline.create(PipelineOptionsFactory.fromArgs(args).withValidation().create());

Schema schema = SchemaBuilder
                .record("table").namespace("org.apache.avro.ipc")
                .fields()
                .name("id").type("int").noDefault()
                .name("number").type("int").noDefault()
                .name("name").type().stringType().noDefault()
                .name("password").type().stringType().noDefault()

p.apply(JdbcIO.<GenericRecord> read()
            .withDataSourceConfiguration(JdbcIO.DataSourceConfiguration.create(
                    "org.postgresql.Driver", "jdbc:postgresql://localhost:port/database")
                    .withUsername("username")
                    .withPassword("password"))
                .withQuery("select * from table")
                .withRowMapper((JdbcIO.RowMapper<GenericRecord>) resultSet -> {
                        GenericRecord record = new GenericData.Record(schema);
                        ResultSetMetaData metadata = resultSet.getMetaData();
                        int columnsNumber = metadata.getColumnCount();
                        for(int i=0; i<columnsNumber; i++) {
                            String columnValue = resultSet.getString(i+1);
                            record.put(i, columnValue);
                        }
                    return record;
                })
                .withCoder(AvroCoder.of(schema)))
            .apply(FileIO.<GenericRecord>write()
                    .via(ParquetIO.sink(schema).withCompressionCodec(CompressionCodecName.SNAPPY))
                    .to("somethingg.parquet")
                    );
p.run()

我得到这个错误

Exception in thread "main" java.lang.IllegalArgumentException: unable to serialize DoFnWithExecutionInformation{doFn=org.apache.beam.sdk.io.jdbc.JdbcIO$ReadFn@4393593c, mainOutputTag=Tag<output>, schemaInformation=DoFnSchemaInformation{elementConverters=[]}}
    at org.apache.beam.sdk.util.SerializableUtils.serializeToByteArray(SerializableUtils.java:55)
    at org.apache.beam.repackaged.beam_runners_direct_java.runners.core.construction.ParDoTranslation.translateDoFn(ParDoTranslation.java:564)
    at org.apache.beam.repackaged.beam_runners_direct_java.runners.core.construction.ParDoTranslation$1.translateDoFn(ParDoTranslation.java:212)
    at org.apache.beam.repackaged.beam_runners_direct_java.runners.core.construction.ParDoTranslation.payloadForParDoLike(ParDoTranslation.java:705)
    at org.apache.beam.repackaged.beam_runners_direct_java.runners.core.construction.ParDoTranslation.translateParDo(ParDoTranslation.java:208)
    at org.apache.beam.repackaged.beam_runners_direct_java.runners.core.construction.ParDoTranslation.translateParDo(ParDoTranslation.java:187)
    at org.apache.beam.repackaged.beam_runners_direct_java.runners.core.construction.ParDoTranslation$ParDoTranslator.translate(ParDoTranslation.java:125)
    at org.apache.beam.repackaged.beam_runners_direct_java.runners.core.construction.PTransformTranslation.toProto(PTransformTranslation.java:155)
    at org.apache.beam.repackaged.beam_runners_direct_java.runners.core.construction.ParDoTranslation.getParDoPayload(ParDoTranslation.java:651)
    at org.apache.beam.repackaged.beam_runners_direct_java.runners.core.construction.ParDoTranslation.isSplittable(ParDoTranslation.java:666)
    at org.apache.beam.repackaged.beam_runners_direct_java.runners.core.construction.PTransformMatchers$6.matches(PTransformMatchers.java:269)
    at org.apache.beam.sdk.Pipeline$2.visitPrimitiveTransform(Pipeline.java:280)
    at org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:665)
    at org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:657)
    at org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:657)
    at org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:657)
    at org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:657)
    at org.apache.beam.sdk.runners.TransformHierarchy$Node.access$600(TransformHierarchy.java:317)
    at org.apache.beam.sdk.runners.TransformHierarchy.visit(TransformHierarchy.java:251)
    at org.apache.beam.sdk.Pipeline.traverseTopologically(Pipeline.java:458)
    at org.apache.beam.sdk.Pipeline.replace(Pipeline.java:258)
    at org.apache.beam.sdk.Pipeline.replaceAll(Pipeline.java:208)
    at org.apache.beam.runners.direct.DirectRunner.run(DirectRunner.java:170)
    at org.apache.beam.runners.direct.DirectRunner.run(DirectRunner.java:67)
    at org.apache.beam.sdk.Pipeline.run(Pipeline.java:313)
    at org.apache.beam.sdk.Pipeline.run(Pipeline.java:299)
    at com.click.example.StarterPipeline.main(StarterPipeline.java:196)
Caused by: java.io.NotSerializableException: org.apache.avro.Schema$RecordSchema
    at java.base/java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1185)
    at java.base/java.io.ObjectOutputStream.writeArray(ObjectOutputStream.java:1379)
    at java.base/java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1175)
    at java.base/java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1553)
    at java.base/java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1510)
    at java.base/java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1433)
    at java.base/java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1179)
    at java.base/java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1553)
    at java.base/java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1510)
    at java.base/java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1433)
    at java.base/java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1179)
    at java.base/java.io.ObjectOutputStream.writeArray(ObjectOutputStream.java:1379)
    at java.base/java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1175)
    at java.base/java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1553)
    at java.base/java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1510)
    at java.base/java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1433)
    at java.base/java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1179)
    at java.base/java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1553)
    at java.base/java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1510)
    at java.base/java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1433)
    at java.base/java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1179)
    at java.base/java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1553)
    at java.base/java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1510)
    at java.base/java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1433)
    at java.base/java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1179)
    at java.base/java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:349)
    at org.apache.beam.sdk.util.SerializableUtils.serializeToByteArray(SerializableUtils.java:51)
    ... 26 more

共1个答案

匿名用户

该错误在堆栈跟踪中得到了很好的解释：由java.io引起。NotSerializableException： org.apache.avro.Schema$RecordSchema。

with RowMapper（）接受一个可序列化的RowMapper

我能想到的解决方法很少：

>

在lambda中创建模式：

 
        在这种情况下，架构实例不会被序列化；
每次调用lambda时都会创建它；

将模式序列化（例如序列化为Json字符串）到lambda之外的可序列化对象，然后在lambda中反序列化它：

 
        它与上面的基本相同，但有一个额外的序列化步骤；
在lambda中，它仍然必须在每次调用时反序列化；

查找/编写可序列化的Schema实现：

 
        可能不可能或很难做到；
可能会像上述方法那样具有更少的开销，因为反序列化只会在创建RowMapper的实例时发生

我认为在lambda中创建模式的新实例是完全可以的，除非它会导致问题。


		      
                相关问题
                

																                
					
										   异步管道会自动取消订阅可观察，如果我改变可观察？
										   异步管道是否从服务中定义并从组件变量指向的可观察对象取消订阅？
										   使用订阅与异步管道时，同一可观察对象的不同结果
										   如果可观察对象使用异步管道，是否需要取消订阅？
										   结合主体时不更新在模板中的异步管道可观察
										   角度观察取消订阅异步管道
										   *具有多个异步管道变量的ngIF
										   在视图中使用@input可观察，但组件中的管道函数被忽略
										   如何将异步管道过滤成Angular子组件
										   角-在可观察

如何创建从postgres到镶木地板的管道？

共1个答案

相关问题

热门标签

微信关注