You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "celltobig (via GitHub)" <gi...@apache.org> on 2023/05/18 10:48:57 UTC
[GitHub] [incubator-paimon] celltobig opened a new issue, #1170: [Bug]
celltobig opened a new issue, #1170:
URL: https://github.com/apache/incubator-paimon/issues/1170
### Search before asking
- [X] I searched in the [issues](https://github.com/apache/incubator-paimon/issues) and found nothing similar.
### Paimon version
0.5
### Compute Engine
flink-1.14.0
### Minimal reproduce step
SET sql-client.execution.result-mode=TABLEAU;
CREATE CATALOG my_hive WITH (
'type' = 'hive',
'hive-conf-dir' = '/opt/conf'
);
CREATE CATALOG my_paimon WITH (
'type' = 'paimon',
'metastore' = 'hive',
'uri'='thrift://xxxxxxx:9083',
'hive-conf-dir'='/opt/conf',
'hadoop-conf-dir'='/etc/hadoop/conf',
'warehouse' = 'hdfs://xxxxx/user/hive/warehouse'
);
USE CATALOG my_paimon;
use test_flink;
SET 'execution.checkpointing.interval' = '10 s';
drop TABLE IF EXISTS `my_hive`.`test_flink`.`ods_kafka_cowell_oms_order_base_stream_v2`;
CREATE TABLE `my_hive`.`test_flink`.`ods_kafka_cowell_oms_order_base_stream_v2` (
`pt` as proctime(),
`id` bigint,
`order_id` bigint,
`business_id` bigint,
`business_name` varchar(128),
`store_id` bigint,
`gmt_create` timestamp(3),
`gmt_update` timestamp(3)
)WITH (
'properties.auto.commit.interval.ms' = '1000',
'canal-json.ignore-parse-errors' = 'true',
'format' = 'canal-json',
'properties.bootstrap.servers' = 'xxxxxxxx',
'connector' = 'kafka',
'topic' = 'cowell_order_order_base',
'scan.startup.mode' = 'group-offsets',
'properties.auto.offset.reset.strategy' = 'earliest',
'properties.group.id' = 'test_flink_group_v2',
'properties.enable.auto.commit' = 'true'
);
-- drop table `my_paimon`.`test_flink`.`order_base_v2`;
CREATE TABLE IF NOT EXISTS `my_paimon`.`test_flink`.`order_base_v2`(
`dt` STRING,
`pt` timestamp,
`ts` timestamp,
`id` bigint ,
`order_id` bigint,
`business_id` bigint,
`business_name` STRING,
`store_id` bigint,
`gmt_create` timestamp,
`gmt_update` timestamp,
PRIMARY KEY (dt, id) NOT ENFORCED
)
PARTITIONED BY (dt)
with (
'bucket'='3',
'bucket-key'='id',
'file.format'='parquet',
'merge-engine'='deduplicate',
'changelog-producer' = 'full-compaction', --指定 changelog-producer 为 full-compaction
'changelog-producer.compaction-interval' = '1 min', --指定 campaction 的间隔为40秒
'snapshot.time-retained' = '1 h',
'snapshot.num-retained.min'='2',
'snapshot.num-retained.max'='3',
'full-compaction.delta-commits' ='1'
);
-- job
INSERT INTO `my_paimon`.`test_flink`.`order_base_v2`
SELECT
DATE_FORMAT(gmt_create,'yyyyMMdd') as dt, -- 以订单生成时间为单位分区
-- proctime() as dt, -- 入湖的时间
proctime() as pt, -- 入湖的时间
gmt_update as ts, -- 业务发生的event_time
`id`,
`order_id`,
`business_id`,
`business_name`,
`store_id`,
`gmt_create`,
`gmt_update`
FROM
`my_hive`.`test_flink`.`ods_kafka_cowell_oms_order_base_stream_v2`
;
### What doesn't meet your expectations?
insert into is error
Exception in thread "main" org.apache.flink.table.client.SqlClientException: Unexpected exception. This is a bug. Please consider filing an issue.
at org.apache.flink.table.client.SqlClient.startClient(SqlClient.java:201)
at org.apache.flink.table.client.SqlClient.main(SqlClient.java:161)
Caused by: java.lang.BootstrapMethodError: java.lang.NoClassDefFoundError: org/apache/flink/util/function/SerializableFunction
at org.apache.paimon.flink.sink.FileStoreSink.createCommitterFactory(FileStoreSink.java:67)
at org.apache.paimon.flink.sink.FlinkSink.sinkFrom(FlinkSink.java:148)
at org.apache.paimon.flink.sink.FlinkSink.sinkFrom(FlinkSink.java:113)
at org.apache.paimon.flink.sink.FlinkSinkBuilder.build(FlinkSinkBuilder.java:100)
at org.apache.paimon.flink.sink.FlinkTableSink.lambda$getSinkRuntimeProvider$0(FlinkTableSink.java:147)
at org.apache.paimon.flink.PaimonDataStreamSinkProvider.consumeDataStream(PaimonDataStreamSinkProvider.java:39)
at org.apache.flink.table.planner.plan.nodes.exec.common.CommonExecSink.applySinkProvider(CommonExecSink.java:294)
at org.apache.flink.table.planner.plan.nodes.exec.common.CommonExecSink.createSinkTransformation(CommonExecSink.java:145)
at org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecSink.translateToPlanInternal(StreamExecSink.java:140)
at org.apache.flink.table.planner.plan.nodes.exec.ExecNodeBase.translateToPlan(ExecNodeBase.java:134)
at org.apache.flink.table.planner.delegation.StreamPlanner$$anonfun$1.apply(StreamPlanner.scala:71)
at org.apache.flink.table.planner.delegation.StreamPlanner$$anonfun$1.apply(StreamPlanner.scala:70)
at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
at scala.collection.Iterator$class.foreach(Iterator.scala:891)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1334)
at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
at scala.collection.TraversableLike$class.map(TraversableLike.scala:234)
at scala.collection.AbstractTraversable.map(Traversable.scala:104)
at org.apache.flink.table.planner.delegation.StreamPlanner.translateToPlan(StreamPlanner.scala:70)
at org.apache.flink.table.planner.delegation.PlannerBase.translate(PlannerBase.scala:185)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.translate(TableEnvironmentImpl.java:1665)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:752)
at org.apache.flink.table.client.gateway.local.LocalExecutor.lambda$executeModifyOperations$4(LocalExecutor.java:222)
at org.apache.flink.table.client.gateway.context.ExecutionContext.wrapClassLoader(ExecutionContext.java:88)
at org.apache.flink.table.client.gateway.local.LocalExecutor.executeModifyOperations(LocalExecutor.java:222)
at org.apache.flink.table.client.cli.CliClient.callInserts(CliClient.java:571)
at org.apache.flink.table.client.cli.CliClient.callInsert(CliClient.java:560)
at org.apache.flink.table.client.cli.CliClient.callOperation(CliClient.java:420)
at org.apache.flink.table.client.cli.CliClient.lambda$executeStatement$1(CliClient.java:332)
at java.util.Optional.ifPresent(Optional.java:159)
at org.apache.flink.table.client.cli.CliClient.executeStatement(CliClient.java:325)
at org.apache.flink.table.client.cli.CliClient.executeFile(CliClient.java:314)
at org.apache.flink.table.client.cli.CliClient.executeInNonInteractiveMode(CliClient.java:230)
at org.apache.flink.table.client.SqlClient.openCli(SqlClient.java:153)
at org.apache.flink.table.client.SqlClient.start(SqlClient.java:95)
at org.apache.flink.table.client.SqlClient.startClient(SqlClient.java:187)
... 1 more
Caused by: java.lang.NoClassDefFoundError: org/apache/flink/util/function/SerializableFunction
... 39 more
Caused by: java.lang.ClassNotFoundException: org.apache.flink.util.function.SerializableFunction
at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
at org.apache.flink.util.FlinkUserCodeClassLoader.loadClassWithoutExceptionHandling(FlinkUserCodeClassLoader.java:64)
at org.apache.flink.util.FlinkUserCodeClassLoader.loadClass(FlinkUserCodeClassLoader.java:48)
at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
... 39 more
### Anything else?
_No response_
### Are you willing to submit a PR?
- [X] I'm willing to submit a PR!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] [Bug] [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi closed issue #1170: [Bug]
URL: https://github.com/apache/paimon/issues/1170
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] JingsongLi commented on issue #1170: [Bug]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on issue #1170:
URL: https://github.com/apache/incubator-paimon/issues/1170#issuecomment-1553917731
I think you are using old Flink 1.14 versions, just like 1.14.1.
We can fix this by: Upgrade your Flink to 1.14.6 or remove all `org.apache.flink.util.function.SerializableFunction` in paimon-flink (can be replaced by `org.apache.paimon.utils.SerializableFunction`).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] celltobig commented on issue #1170: [Bug]
Posted by "celltobig (via GitHub)" <gi...@apache.org>.
celltobig commented on issue #1170:
URL: https://github.com/apache/incubator-paimon/issues/1170#issuecomment-1560391001
I find the checkpoint error's reason ， I set checkpoint timeout is too short
but ,this job work is when kafka is error ,
may be I use kafka-client version is not true
kafka-client:3.0.0
flink-connector-kafka_2.12-1.14.6.jar
now ,I change kafka-client:3.0.0 to kafka-clients-2.1.1 try
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] celltobig commented on issue #1170: [Bug]
Posted by "celltobig (via GitHub)" <gi...@apache.org>.
celltobig commented on issue #1170:
URL: https://github.com/apache/incubator-paimon/issues/1170#issuecomment-1554312731
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] celltobig commented on issue #1170: [Bug]
Posted by "celltobig (via GitHub)" <gi...@apache.org>.
celltobig commented on issue #1170:
URL: https://github.com/apache/incubator-paimon/issues/1170#issuecomment-1554311730
use Flink to 1.14.6 submit OK ,but checkpoint not work well
like this
USE CATALOG my_paimon;
use test_flink;
SET 'execution.checkpointing.interval' = '60 s';
drop TABLE IF EXISTS `my_hive`.`test_flink`.`ods_kafka_cowell_oms_order_base_stream_v2`;
CREATE TABLE `my_hive`.`test_flink`.`ods_kafka_cowell_oms_order_base_stream_v2` (
`pt` as proctime(),
`id` bigint,
`order_id` bigint,
`business_id` bigint,
`business_name` varchar(128),
`store_id` bigint,
`gmt_create` timestamp(3),
`gmt_update` timestamp(3)
)WITH (
'properties.auto.commit.interval.ms' = '1000',
'canal-json.ignore-parse-errors' = 'true',
'format' = 'canal-json',
'properties.bootstrap.servers' = 'xxxxxxx:9092',
'connector' = 'kafka',
'topic' = 'cowell_order_order_base',
'scan.startup.mode' = 'group-offsets',
'properties.auto.offset.reset.strategy' = 'earliest',
'properties.group.id' = 'test_flink_group_v2',
'properties.enable.auto.commit' = 'true'
);
CREATE TABLE IF NOT EXISTS `my_paimon`.`test_flink`.`order_base_v2`(
`dt` STRING,
`pt` timestamp,
`ts` timestamp,
`id` bigint ,
`order_id` bigint,
`business_id` bigint,
`business_name` STRING,
`store_id` bigint,
`gmt_create` timestamp,
`gmt_update` timestamp,
PRIMARY KEY (dt, id) NOT ENFORCED
)
PARTITIONED BY (dt)
with (
'bucket'='3',
'bucket-key'='id',
'file.format'='parquet',
'merge-engine'='deduplicate',
'changelog-producer' = 'full-compaction', --指定 changelog-producer 为 full-compaction
'full-compaction.delta-commits' ='30'
);
-- job
INSERT INTO `my_paimon`.`test_flink`.`order_base_v2`
SELECT
DATE_FORMAT(gmt_create,'yyyyMMdd') as dt, -- 以订单生成时间为单位分区
-- proctime() as dt, -- 入湖的时间
proctime() as pt, -- 入湖的时间
gmt_update as ts, -- 业务发生的event_time
`id`,
`order_id`,
`business_id`,
`business_name`,
`store_id`,
`gmt_create`,
`gmt_update`
FROM
`my_hive`.`test_flink`.`ods_kafka_cowell_oms_order_base_stream_v2`
;
-- select * from `my_paimon`.`test_flink`.`order_base_v2`;
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-paimon] reswqa commented on issue #1170: [Bug]
Posted by "reswqa (via GitHub)" <gi...@apache.org>.
reswqa commented on issue #1170:
URL: https://github.com/apache/incubator-paimon/issues/1170#issuecomment-1555874843
> remove all org.apache.flink.util.function.SerializableFunction in paimon-flink (can be replaced by org.apache.paimon.utils.SerializableFunction).
That's make sense, I will open a PR to do this fix.
> use Flink to 1.14.6 submit OK ,but checkpoint not work well
Can you find the root cause of the failed async checkpoint in the `jobmanager` log?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org