apache beam的clickhouseio中的数组类型(数据流)

vdgimpew  于 2021-07-15  发布在  ClickHouse
关注(0)|答案(0)|浏览(393)

我使用apachebeam使用json并插入clickhouse。
当前数组数据类型有问题。
在添加数组类型的字段之前，一切都正常

  1. Schema.Field.of("inputs.value", Schema.FieldType.array(Schema.FieldType.INT64).withNullable(true))

转换代码

  1. p.apply(transformNameSuffix + "ReadFromPubSub",
  2.         PubsubIO.readStrings().fromSubscription(chainConfig.getPubSubSubscriptionPrefix() + "transactions").withIdAttribute(PUBSUB_ID_ATTRIBUTE))
  3.         .apply(transformNameSuffix + "ReadFromPubSub", ParDo.of(new DoFn<String, Row>() {
  4.             @ProcessElement
  5.             public void processElement(ProcessContext c) {
  6.                 String item = c.element();
  8.                 //System.out.print(item);
  9.                 Transaction transaction = JsonUtils.parseJson(item, Transaction.class);
  10.                 c.output(Row.withSchema(Schemas.TRANSACTIONS)
  11.                         .addValues(*****,
  12.                                  *****
  13.                                    .......
  15.                 transaction.getInputValues()).build());}
  17.         })).setRowSchema(Schemas.TRANSACTIONS).apply(
  18.         ClickHouseIO.<Row>write(
  19.                 chainConfig.getClickhouseJDBCURI(),
  20.                 chainConfig.getTransactionsTable())
  21.                 .withMaxRetries(3)
  22.                 .withMaxInsertBlockSize(1)
  23.                 .withInitialBackoff(Duration.standardSeconds(5))
  24.                 .withInsertDeduplicate(true)
  25.                 .withInsertDistributedSync(false));

生成输入的方法

  1. public List<Long> getInputValues() {
  2.     List<Long> values = Lists.newArrayList();
  4.     for (TransactionInput eachInput : inputs) {
  5.         System.out.print(eachInput.getValue());
  6.         values.add(eachInput.getValue());
  7.     }
  9.     return values;
  10. }

我得到的错误是：

  1. ru.yandex.clickhouse.except.ClickHouseException: ClickHouse exception, code: 33, host: 35.202.46.77, port: 8123; Code: 33, e.displayText() = DB::Exception: Cannot read all data. Bytes read: 6. Bytes expected: 15. (version 19.17.4.11 (official build))
  3.     at ru.yandex.clickhouse.except.ClickHouseExceptionSpecifier.specify(ClickHouseExceptionSpecifier.java:58)
  4.     at ru.yandex.clickhouse.except.ClickHouseExceptionSpecifier.specify(ClickHouseExceptionSpecifier.java:28)
  5.     at ru.yandex.clickhouse.ClickHouseStatementImpl.checkForErrorAndThrow(ClickHouseStatementImpl.java:875)
  6.     at ru.yandex.clickhouse.ClickHouseStatementImpl.sendStream(ClickHouseStatementImpl.java:851)
  7.     at ru.yandex.clickhouse.Writer.send(Writer.java:106)
  8.     at ru.yandex.clickhouse.Writer.send(Writer.java:141)
  9.     at ru.yandex.clickhouse.ClickHouseStatementImpl.sendRowBinaryStream(ClickHouseStatementImpl.java:764)
  10.     at ru.yandex.clickhouse.ClickHouseStatementImpl.sendRowBinaryStream(ClickHouseStatementImpl.java:758)
  11.     at org.apache.beam.sdk.io.clickhouse.ClickHouseIO$WriteFn.flush(ClickHouseIO.java:427)
  12.     at org.apache.beam.sdk.io.clickhouse.ClickHouseIO$WriteFn.processElement(ClickHouseIO.java:411)
  13.     at org.apache.beam.sdk.io.clickhouse.AutoValue_ClickHouseIO_WriteFn$DoFnInvoker.invokeProcessElement(Unknown Source)
  14.     at org.apache.beam.repackaged.direct_java.runners.core.SimpleDoFnRunner.invokeProcessElement(SimpleDoFnRunner.java:222)
  15.     at org.apache.beam.repackaged.direct_java.runners.core.SimpleDoFnRunner.processElement(SimpleDoFnRunner.java:183)
  16.     at org.apache.beam.repackaged.direct_java.runners.core.SimplePushbackSideInputDoFnRunner.processElementInReadyWindows(SimplePushbackSideInputDoFnRunner.java:78)
  17.     at org.apache.beam.runners.direct.ParDoEvaluator.processElement(ParDoEvaluator.java:216)
  18.     at org.apache.beam.runners.direct.DoFnLifecycleManagerRemovingTransformEvaluator.processElement(DoFnLifecycleManagerRemovingTransformEvaluator.java:54)
  19.     at org.apache.beam.runners.direct.DirectTransformExecutor.processElements(DirectTransformExecutor.java:160)
  20.     at org.apache.beam.runners.direct.DirectTransformExecutor.run(DirectTransformExecutor.java:124)
  21.     at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
  22.     at java.util.concurrent.FutureTask.run(FutureTask.java:266)
  23.     at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
  24.     at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
  25.     at java.lang.Thread.run(Thread.java:748)
  26. Caused by: java.lang.Throwable: Code: 33, e.displayText() = DB::Exception: Cannot read all data. Bytes read: 6. Bytes expected: 15. (version 19.17.4.11 (official build))
  28.     at ru.yandex.clickhouse.except.ClickHouseExceptionSpecifier.specify(ClickHouseExceptionSpecifier.java:53)
  29.     ... 22 more
  31. Feb 06, 2020 9:04:38 PM org.apache.beam.sdk.io.clickhouse.ClickHouseIO$WriteFn flush
  32. WARNING: Error writing to ClickHouse. Retry attempt[1]
  33. ru.yandex.clickhouse.except.ClickHouseException: ClickHouse exception, code: 33, host: 35.202.46.77, port: 8123; Code: 33, e.displayText() = DB::Exception: Cannot read all data. Bytes read: 6. Bytes expected: 93. (version 19.17.4.11 (official build))
  35.     at ru.yandex.clickhouse.except.ClickHouseExceptionSpecifier.specify(ClickHouseExceptionSpecifier.java:58)
  36.     at ru.yandex.clickhouse.except.ClickHouseExceptionSpecifier.specify(ClickHouseExceptionSpecifier.java:28)
  37.     at ru.yandex.clickhouse.ClickHouseStatementImpl.checkForErrorAndThrow(ClickHouseStatementImpl.java:875)
  38.     at ru.yandex.clickhouse.ClickHouseStatementImpl.sendStream(ClickHouseStatementImpl.java:851)
  39.     at ru.yandex.clickhouse.Writer.send(Writer.java:106)
  40.     at ru.yandex.clickhouse.Writer.send(Writer.java:141)
  41.     at ru.yandex.clickhouse.ClickHouseStatementImpl.sendRowBinaryStream(ClickHouseStatementImpl.java:764)
  42.     at ru.yandex.clickhouse.ClickHouseStatementImpl.sendRowBinaryStream(ClickHouseStatementImpl.java:758)
  43.     at org.apache.beam.sdk.io.clickhouse.ClickHouseIO$WriteFn.flush(ClickHouseIO.java:427)
  44.     at org.apache.beam.sdk.io.clickhouse.ClickHouseIO$WriteFn.processElement(ClickHouseIO.java:411)
  45.     at org.apache.beam.sdk.io.clickhouse.AutoValue_ClickHouseIO_WriteFn$DoFnInvoker.invokeProcessElement(Unknown Source)
  46.     at org.apache.beam.repackaged.direct_java.runners.core.SimpleDoFnRunner.invokeProcessElement(SimpleDoFnRunner.java:222)
  47.     at org.apache.beam.repackaged.direct_java.runners.core.SimpleDoFnRunner.processElement(SimpleDoFnRunner.java:183)
  48.     at org.apache.beam.repackaged.direct_java.runners.core.SimplePushbackSideInputDoFnRunner.processElementInReadyWindows(SimplePushbackSideInputDoFnRunner.java:78)
  49.     at org.apache.beam.runners.direct.ParDoEvaluator.processElement(ParDoEvaluator.java:216)
  50.     at org.apache.beam.runners.direct.DoFnLifecycleManagerRemovingTransformEvaluator.processElement(DoFnLifecycleManagerRemovingTransformEvaluator.java:54)
  51.     at org.apache.beam.runners.direct.DirectTransformExecutor.processElements(DirectTransformExecutor.java:160)
  52.     at org.apache.beam.runners.direct.DirectTransformExecutor.run(DirectTransformExecutor.java:124)
  53.     at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
  54.     at java.util.concurrent.FutureTask.run(FutureTask.java:266)
  55.     at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
  56.     at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
  57.     at java.lang.Thread.run(Thread.java:748)
  58. Caused by: java.lang.Throwable: Code: 33, e.displayText() = DB::Exception: Cannot read all data. Bytes read: 6. Bytes expected: 93. (version 19.17.4.11 (official build))
  60.     at ru.yandex.clickhouse.except.ClickHouseExceptionSpecifier.specify(ClickHouseExceptionSpecifier.java:53)
  61.     ... 22 more
  63. Feb 06, 2020 9:04:39 PM org.apache.beam.sdk.io.clickhouse.ClickHouseIO$WriteFn flush
  64. WARNING: Error writing to ClickHouse. Retry attempt[1]
  65. ru.yandex.clickhouse.except.ClickHouseException: ClickHouse exception, code: 33, host: 35.202.46.77, port: 8123; Code: 33, e.displayText() = DB::Exception: Cannot read all data. Bytes read: 5. Bytes expected: 2641. (version 19.17.4.11 (official build)

clikhouse架构：

  1. CREATE TABLE IF NOT EXISTS transactions_streaming_small ( 
  3. *****, 
  4. *****, 
  6.   inputs Nested ( value Nullable(UInt64) ) ) 
  7. ENGINE = MergeTree() PARTITION BY toYYYYMM(block_date_time)

有什么问题？
[clickhouse版本19.17.4.11（官方版本）]

clickhouseapache-beamgoogle-cloud-dataflowdataflow

暂无答案！

目前还没有任何答案，快来回答吧！

我来回答

相关问题

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新问答

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com