BEAM：ORACLE 连接器：NUMBER 数据类型：非法参数类型异常答案

【问题标题】：BEAM :ORACLE connector: NUMBER Data type : illegal argument type exceptionBEAM：ORACLE 连接器：NUMBER 数据类型：非法参数类型异常
【发布时间】：2021-08-10 04:08:00
【问题描述】：

String sql = "select CUSTOMER_ID, CITY, CUSTOMER_FIRSTTNAME from LIMA.CHECK_FLOW where CUSTOMER_ID < 1000";

        Schema schema = Schema.of(
        Schema.Field.of("CUSTOMER_ID", Schema.FieldType.INT32),
        Schema.Field.of("CITY", Schema.FieldType.STRING),
        Schema.Field.of("CUSTOMER_FIRSTTNAME", Schema.FieldType.STRING)
);

PCollection<Row> result = pipe.apply(JdbcIO.readRows()
    .withDataSourceConfiguration(configuration)
    .withQuery(sql)
    .withFetchSize(1000)
).setCoder(RowCoder.of(schema)).setRowSchema(schema);

PipelineResult nRes = pipe.run();

我们正在测试 Beams，看看它是否适合我们的用例。上面是一个简单的例子。我收到以下错误

Caused by: org.apache.beam.sdk.util.UserCodeException: java.lang.IllegalArgumentException
    at org.apache.beam.sdk.util.UserCodeException.wrap(UserCodeException.java:39)
    at org.apache.beam.sdk.io.jdbc.JdbcIO$ReadFn$DoFnInvoker.invokeProcessElement(Unknown Source)
    at org.apache.beam.runners.core.SimpleDoFnRunner.invokeProcessElement(SimpleDoFnRunner.java:232)
    at org.apache.beam.runners.core.SimpleDoFnRunner.processElement(SimpleDoFnRunner.java:191)
    at org.apache.beam.runners.flink.metrics.DoFnRunnerWithMetricsUpdate.processElement(DoFnRunnerWithMetricsUpdate.java:62)
    at org.apache.beam.runners.flink.translation.functions.FlinkDoFnFunction.flatMap(FlinkDoFnFunction.java:122)
    at org.apache.beam.runners.flink.translation.functions.FlinkDoFnFunction.flatMap(FlinkDoFnFunction.java:59)
    at org.apache.flink.runtime.operators.chaining.ChainedFlatMapDriver.collect(ChainedFlatMapDriver.java:79)
    at org.apache.flink.runtime.operators.util.metrics.CountingCollector.collect(CountingCollector.java:35)
    at org.apache.flink.runtime.operators.DataSourceTask.invoke(DataSourceTask.java:208)
    at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:779)
    at org.apache.flink.runtime.taskmanager.Task.run(Task.java:566)
    at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.IllegalArgumentException
    at org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions.checkArgument(Preconditions.java:127)
    at org.apache.beam.sdk.io.jdbc.LogicalTypes$FixedPrecisionNumeric.toInputType(LogicalTypes.java:268)
    at org.apache.beam.sdk.io.jdbc.LogicalTypes$FixedPrecisionNumeric.toInputType(LogicalTypes.java:246)
    at org.apache.beam.sdk.io.jdbc.SchemaUtil.lambda$createLogicalTypeExtractor$ca0ab2ec$1(SchemaUtil.java:279)
    at org.apache.beam.sdk.io.jdbc.SchemaUtil$BeamRowMapper.mapRow(SchemaUtil.java:344)
    at org.apache.beam.sdk.io.jdbc.SchemaUtil$BeamRowMapper.mapRow(SchemaUtil.java:322)
    at org.apache.beam.sdk.io.jdbc.JdbcIO$ReadFn.processElement(JdbcIO.java:924)

任何解决错误的帮助将不胜感激。

【问题讨论】：

试过了，将 customer_id 的数据类型更改为整数、小数等。但是，我得到了同样的异常。
String sql = "从 LIMA.CHECK_FLOW 中选择 CUSTOMER_ID、CITY、CUSTOMER_FIRSTTNAME，其中 CUSTOMER_ID
数据库中 CUSTOMER_ID 的实际类型是什么？
NUMBER ... 甲骨文
想知道这是否与此问题相同。 issues.apache.org/jira/browse/…

标签： java apache-flink apache-beam

【解决方案1】：

我遇到了同样的问题，并作为this pull request 的一部分得到了解决。

该版本将在 2022 年 1 月左右发布。现在，您可以通过创建 org/apache/beam/sdk/io/jdbc/LogicalTypes.java 类并复制以下内容来将此 PR 的更改应用到您自己的项目中这个file 在吗？我也有同样的建议。

【讨论】：