You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@seatunnel.apache.org by GitBox <gi...@apache.org> on 2022/08/31 16:14:13 UTC
[GitHub] [incubator-seatunnel] hk-lrzy opened a new pull request, #2586: [SeaTunnel][Connector-v2][Doris]Support Doris.
hk-lrzy opened a new pull request, #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586
<!--
Thank you for contributing to SeaTunnel! Please make sure that your code changes
are covered with tests. And in case of new features or big changes
remember to adjust the documentation.
Feel free to ping committers for the review!
## Contribution Checklist
- Make sure that the pull request corresponds to a [GITHUB issue](https://github.com/apache/incubator-seatunnel/issues).
- Name the pull request in the form "[Feature] [component] Title of the pull request", where *Feature* can be replaced by `Hotfix`, `Bug`, etc.
- Minor fixes should be named following this pattern: `[hotfix] [docs] Fix typo in README.md doc`.
-->
## Purpose of this pull request
<!-- Describe the purpose of this pull request. For example: This pull request adds checkstyle plugin.-->
## Check list
* [ ] Code changed are covered with tests, or it does not need tests for reason:
* [ ] If any new Jar binary package adding in your PR, please add License Notice according
[New License Guide](https://github.com/apache/incubator-seatunnel/blob/dev/docs/en/contribution/new-license.md)
* [ ] If necessary, please update the documentation to describe the new feature. https://github.com/apache/incubator-seatunnel/tree/dev/docs
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] CalvinKirs commented on pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris.
Posted by GitBox <gi...@apache.org>.
CalvinKirs commented on PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#issuecomment-1233598867
@xtr1993 Hi, PTAL,thx
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] hk-lrzy commented on a diff in pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
hk-lrzy commented on code in PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#discussion_r964980512
##########
seatunnel-e2e/seatunnel-spark-connector-v2-e2e/connector-spark-e2e-base/src/test/java/org/apache/seatunnel/e2e/spark/SparkContainer.java:
##########
@@ -63,7 +63,7 @@ public abstract class SparkContainer {
private static final int WAIT_SPARK_JOB_SUBMIT = 5000;
@BeforeEach
- public void before() {
+ public void before() throws InterruptedException {
Review Comment:
sure
##########
seatunnel-e2e/seatunnel-spark-connector-v2-e2e/connector-doris-spark-e2e/src/test/resources/doris/fakesource_to_doris.conf:
##########
@@ -0,0 +1,64 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+######
+###### This config file is a demonstration of streaming processing in seatunnel config
+######
+
+env {
+ # You can set flink configuration here
+ execution.parallelism = 1
+ job.mode = "BATCH"
+ #execution.checkpoint.interval = 10000
+ #execution.checkpoint.data-uri = "hdfs://localhost:9000/checkpoint"
+}
+
+source {
+ # This is a example source plugin **only for test and demonstrate the feature source plugin**
+ FakeSource {
+ result_table_name = "fake"
+ schema = {
+ fields {
+ name = "string"
Review Comment:
sure.
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] hk-lrzy commented on a diff in pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
hk-lrzy commented on code in PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#discussion_r970827723
##########
seatunnel-e2e/seatunnel-spark-connector-v2-e2e/connector-doris-spark-e2e/src/test/resources/doris/fakesource_to_doris.conf:
##########
@@ -0,0 +1,75 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+######
+###### This config file is a demonstration of streaming processing in seatunnel config
+######
+
+env {
+ # You can set flink configuration here
+ job.mode = "BATCH"
+ spark.app.name = "SeaTunnel"
+ spark.executor.instances = 2
+ spark.executor.cores = 1
+ spark.executor.memory = "1g"
+ spark.master = local
+ #execution.checkpoint.interval = 10000
+ #execution.checkpoint.data-uri = "hdfs://localhost:9000/checkpoint"
+}
+
+source {
+ # This is a example source plugin **only for test and demonstrate the feature source plugin**
+ FakeSource {
+ result_table_name = "fake"
+ schema = {
+ fields {
+ user_id = "bigint",
+ city = "string",
+ age = "smallint",
+ sex = "smallint",
+ cost = "bigint",
+ max_dwell_time = "int",
+ min_dwell_time = "int",
+ price = "double"
Review Comment:
Spark can't do this right now, and i think jdbc doesn't test all datatypes in spark e2e?
https://github.com/apache/incubator-seatunnel/pull/2637
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] hk-lrzy commented on a diff in pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
hk-lrzy commented on code in PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#discussion_r970828564
##########
seatunnel-e2e/seatunnel-spark-connector-v2-e2e/connector-doris-spark-e2e/src/test/resources/doris/fakesource_to_doris.conf:
##########
@@ -0,0 +1,75 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+######
+###### This config file is a demonstration of streaming processing in seatunnel config
+######
+
+env {
+ # You can set flink configuration here
Review Comment:
Sorry for mistake your previous comment, i will fixed this issue.
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] hk-lrzy commented on pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
hk-lrzy commented on PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#issuecomment-1246913732
@hailin0 I reply the comments.
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] hk-lrzy commented on a diff in pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
hk-lrzy commented on code in PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#discussion_r969116151
##########
seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/common/HttpClient.java:
##########
@@ -0,0 +1,153 @@
+/*
Review Comment:
Just like `seatunnel-formats`, we will have `seatuennl-client` as parent module and `seatunnel-client-http` as child module
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] hk-lrzy commented on a diff in pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
hk-lrzy commented on code in PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#discussion_r970586989
##########
seatunnel-e2e/seatunnel-spark-connector-v2-e2e/connector-doris-spark-e2e/src/test/resources/doris/fakesource_to_doris.conf:
##########
@@ -0,0 +1,71 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+######
+###### This config file is a demonstration of streaming processing in seatunnel config
+######
+
+env {
Review Comment:
Can you tell more detail about this one. i checked others and env configuration is same like this file.
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] TaoZex commented on a diff in pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
TaoZex commented on code in PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#discussion_r1004118172
##########
seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/sink/loader/DorisStreamLoader.java:
##########
@@ -0,0 +1,129 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.doris.sink.loader;
+
+import static org.apache.seatunnel.connectors.doris.common.DorisConstants.DORIS_SUCCESS_STATUS;
Review Comment:
How about use eum?
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] EricJoy2048 closed pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
EricJoy2048 closed pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
URL: https://github.com/apache/incubator-seatunnel/pull/2586
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] EricJoy2048 commented on a diff in pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
EricJoy2048 commented on code in PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#discussion_r970282322
##########
seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/common/HttpClient.java:
##########
@@ -0,0 +1,153 @@
+/*
Review Comment:
> Just like `seatunnel-formats`, we will have `seatuennl-client` as parent module and `seatunnel-client-http` as child module
`seatunnel-client` is easily misunderstood. I think use `http-connector-base` is better. However, I suggest that we can iterate it in the feature.
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] hk-lrzy commented on a diff in pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
hk-lrzy commented on code in PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#discussion_r970832113
##########
seatunnel-e2e/seatunnel-flink-connector-v2-e2e/connector-doris-flink-e2e/src/test/java/org/apache/seatunnel/e2e/flink/v2/doris/FakeSourceToDorisIT.java:
##########
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.e2e.flink.v2.doris;
+
+import org.apache.seatunnel.e2e.flink.FlinkContainer;
+
+import com.google.common.collect.Lists;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.Container;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.containers.wait.strategy.HostPortWaitStrategy;
+import org.testcontainers.lifecycle.Startables;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
+
+public class FakeSourceToDorisIT extends FlinkContainer {
+ private static final Logger LOG = LoggerFactory.getLogger(FakeSourceToDorisIT.class);
+
+ private static final String DORIS_DRIVER = "com.mysql.cj.jdbc.Driver";
+ private static final String DORIS_CONNECTION_URL = "jdbc:mysql://localhost:9030?rewriteBatchedStatements=true";
+ private static final String DORIS_PASSWD = "";
+ private static final String DORIS_USERNAME = "root";
+
+ private static final String DORIS_DATABASE = "test";
+ private static final String DORIS_TABLE = "seatunnel";
+ private static final String DORIS_DATABASE_DDL = "CREATE DATABASE IF NOT EXISTS `" + DORIS_DATABASE + "`";
+ private static final String DORIS_USE_DATABASE = "USE `" + DORIS_DATABASE + "`";
+ private static final String DORIS_TABLE_DDL = "CREATE TABLE IF NOT EXISTS `" + DORIS_DATABASE + "`.`" + DORIS_TABLE + "` ( " +
+ " `user_id` LARGEINT NOT NULL COMMENT 'id'," +
+ " `date` DATE NOT NULL COMMENT 'date'," +
+ " `city` VARCHAR(20) COMMENT 'city'," +
+ " `age` SMALLINT COMMENT 'age'," +
+ " `sex` TINYINT COMMENT 'sec'," +
+ " `last_visit_date` DATETIME REPLACE DEFAULT '1970-01-01 00:00:00' ," +
+ " `cost` BIGINT SUM DEFAULT '0' ," +
+ " `max_dwell_time` INT MAX DEFAULT '0' ," +
+ " `min_dwell_time` INT MIN DEFAULT '99999'" +
+ ") AGGREGATE KEY(`user_id`, `date`, `city`, `age`, `sex`) DISTRIBUTED BY HASH(`user_id`) BUCKETS 1 PROPERTIES (" +
+ " 'replication_allocation' = 'tag.location.default: 1'" +
+ ");";
+
+ private static final String DORIS_TRUNCATE_TABLE = "TRUNCATE TABLE `" + DORIS_TABLE + "`";
+ private static final String DORIS_SELECT_TABLE = "SELECT COUNT(*) FROM `" + DORIS_TABLE + "`";
+
+ //thanks zhaomin1432 provided the doris images.
+ private static final String DORIS_IMAGE_NAME = "zhaomin1423/doris:1.0.0-b2";
Review Comment:
Thanks, i am already try to build image by myself, and it will push to dockerhub.
Hi, are there any issues with existing doris images on docker hub? On 09/1/2022 @.***> wrote：
it's just use for build-env, you may see the image's tag like [build-env-for-1.1.0](https://hub.docker.com/layers/apache/doris/build-env-for-1.1.0/images/sha256-174d1f3106c200625155416d16d1db02939ce3f976808b995e7c381d939209ae?context=explore)
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] hk-lrzy commented on a diff in pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
hk-lrzy commented on code in PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#discussion_r970608618
##########
seatunnel-connectors-v2/connector-doris/pom.xml:
##########
@@ -0,0 +1,95 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You under the Apache License, Version 2.0
+ (the "License"); you may not use this file except in compliance with
+ the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <parent>
+ <artifactId>seatunnel-connectors-v2</artifactId>
+ <groupId>org.apache.seatunnel</groupId>
+ <version>${revision}</version>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <artifactId>connector-doris</artifactId>
+ <properties>
+ <httpclient.version>4.5.13</httpclient.version>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>connector-common</artifactId>
+ <version>${revision}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>seatunnel-format-json</artifactId>
+ <version>${revision}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.httpcomponents</groupId>
+ <artifactId>httpclient</artifactId>
+ <version>${httpclient.version}</version>
+ </dependency>
+ </dependencies>
+
+ <build>
Review Comment:
<img width="731" alt="截屏2022-09-14 18 12 46" src="https://user-images.githubusercontent.com/14176134/190127261-81396491-f7da-4fe7-ae26-bc3eaeda5cc7.png">
<img width="880" alt="截屏2022-09-14 18 12 50" src="https://user-images.githubusercontent.com/14176134/190127283-3bfaced7-7e24-43d7-a5f0-819f559c51b0.png">
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] hk-lrzy commented on a diff in pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
hk-lrzy commented on code in PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#discussion_r970608988
##########
seatunnel-connectors-v2/connector-doris/pom.xml:
##########
@@ -0,0 +1,95 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You under the Apache License, Version 2.0
+ (the "License"); you may not use this file except in compliance with
+ the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <parent>
+ <artifactId>seatunnel-connectors-v2</artifactId>
+ <groupId>org.apache.seatunnel</groupId>
+ <version>${revision}</version>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <artifactId>connector-doris</artifactId>
+ <properties>
+ <httpclient.version>4.5.13</httpclient.version>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>connector-common</artifactId>
+ <version>${revision}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>seatunnel-format-json</artifactId>
+ <version>${revision}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.httpcomponents</groupId>
+ <artifactId>httpclient</artifactId>
+ <version>${httpclient.version}</version>
+ </dependency>
+ </dependencies>
+
+ <build>
Review Comment:
http client's version must bigger than 4.5.10.
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] TaoZex commented on a diff in pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
TaoZex commented on code in PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#discussion_r1002660997
##########
seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/sink/DorisSink.java:
##########
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.doris.sink;
+
+import org.apache.seatunnel.api.common.PrepareFailException;
+import org.apache.seatunnel.api.sink.SeaTunnelSink;
+import org.apache.seatunnel.api.sink.SinkWriter;
+import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import org.apache.seatunnel.connectors.seatunnel.common.sink.AbstractSimpleSink;
+import org.apache.seatunnel.connectors.seatunnel.common.sink.AbstractSinkWriter;
+
+import org.apache.seatunnel.shade.com.typesafe.config.Config;
+
+import com.google.auto.service.AutoService;
+
+import java.io.IOException;
+
+@AutoService(SeaTunnelSink.class)
+public class DorisSink extends AbstractSimpleSink<SeaTunnelRow, Void> {
+ private static final String DORIS_NAME = "Doris";
+
+ private SeaTunnelRowType seaTunnelRowType;
+ private Config pluginConfig;
+
+ @Override
+ public String getPluginName() {
+ return DORIS_NAME;
+ }
+
+ @Override
+ public void prepare(Config pluginConfig) throws PrepareFailException {
+ this.pluginConfig = pluginConfig;
Review Comment:
Need parameter verification
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] CalvinKirs commented on pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris.
Posted by GitBox <gi...@apache.org>.
CalvinKirs commented on PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#issuecomment-1233845191
This situation (because the framework itself is not integrated) requires us to create it manually. FYI : https://www.testcontainers.org/features/creating_images/
On 09/1/2022 ***@***.***> wrote：
flink
Sure, i am try to , but there have no avaliable docker image for doris now.
—
Reply to this email directly, view it on GitHub, or unsubscribe.
You are receiving this because you commented.Message ID: ***@***.***>
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] EricJoy2048 commented on pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris.
Posted by GitBox <gi...@apache.org>.
EricJoy2048 commented on PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#issuecomment-1235097474
Can you update the title of this pr and add `Sink`. Because I found this is a `sink` connector.
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] hk-lrzy commented on a diff in pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
hk-lrzy commented on code in PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#discussion_r970822670
##########
seatunnel-e2e/seatunnel-spark-connector-v2-e2e/connector-doris-spark-e2e/src/test/java/org/apache/seatunnel/e2e/spark/v2/doris/FakeSourceToDorisIT.java:
##########
@@ -0,0 +1,150 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.e2e.spark.v2.doris;
+
+import org.apache.seatunnel.e2e.spark.SparkContainer;
+
+import org.apache.commons.compress.utils.Lists;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.Container;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.containers.wait.strategy.HostPortWaitStrategy;
+import org.testcontainers.lifecycle.Startables;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
+
+public class FakeSourceToDorisIT extends SparkContainer {
+ private static final Logger LOG = LoggerFactory.getLogger(FakeSourceToDorisIT.class);
+
+ private static final String DORIS_DRIVER = "com.mysql.cj.jdbc.Driver";
+ private static final String DORIS_CONNECTION_URL = "jdbc:mysql://localhost:9030?rewriteBatchedStatements=true";
+ private static final String DORIS_PASSWD = "";
+ private static final String DORIS_USERNAME = "root";
+
+ private static final String DORIS_DATABASE = "test";
+ private static final String DORIS_TABLE = "seatunnel";
+ private static final String DORIS_DATABASE_DDL = "CREATE DATABASE IF NOT EXISTS `" + DORIS_DATABASE + "`";
+ private static final String DORIS_USE_DATABASE = "USE `" + DORIS_DATABASE + "`";
+ private static final String DORIS_TABLE_DDL = "CREATE TABLE IF NOT EXISTS `" + DORIS_DATABASE + "`.`" + DORIS_TABLE + "` ( " +
+ " `user_id` LARGEINT NOT NULL COMMENT 'id'," +
+ " `city` VARCHAR(20) COMMENT 'city'," +
+ " `age` SMALLINT COMMENT 'age'," +
+ " `sex` TINYINT COMMENT 'sec'," +
+ " `cost` BIGINT SUM DEFAULT '0' ," +
+ " `max_dwell_time` INT MAX DEFAULT '0' ," +
+ " `min_dwell_time` INT MIN DEFAULT '99999'" +
+ ") AGGREGATE KEY(`user_id`, `city`, `age`, `sex`) DISTRIBUTED BY HASH(`user_id`) BUCKETS 1 PROPERTIES (" +
+ " 'replication_allocation' = 'tag.location.default: 1'" +
+ ");";
+
+ private static final String DORIS_TABLE_TRUNCATE_TABLE = "TRUNCATE TABLE `" + DORIS_TABLE + "`";
+ private static final String DORIS_SELECT_TABLE = "SELECT COUNT(*) FROM `" + DORIS_TABLE + "`";
+
+ //thanks zhaomin1432 provided the doris images.
+ private static final String DORIS_IMAGE_NAME = "zhaomin1423/doris:1.0.0-b2";
+ private static final int DORIS_FE_PORT = 8030;
+ private static final int DORIS_QUERY_PORT = 8040;
+ private static final int DORIS_BE_PORT = 9030;
+
+ private GenericContainer<?> dorisStandaloneServer;
+ private Connection connection;
+
+ @BeforeEach
+ public void beforeInDoris() throws InterruptedException {
+ super.before();
+ dorisStandaloneServer = new GenericContainer<>(DORIS_IMAGE_NAME)
+ .withNetwork(NETWORK)
+ .withNetworkAliases("seatunnel-doris-network")
+ .withLogConsumer(new Slf4jLogConsumer(LOG));
+ List<String> portBindings = Lists.newArrayList();
+ portBindings.add(String.format("%s:%s", DORIS_FE_PORT, DORIS_FE_PORT));
+ portBindings.add(String.format("%s:%s", DORIS_QUERY_PORT, DORIS_QUERY_PORT));
+ portBindings.add(String.format("%s:%s", DORIS_BE_PORT, DORIS_BE_PORT));
+ dorisStandaloneServer.setPortBindings(portBindings);
+ Startables.deepStart(Stream.of(dorisStandaloneServer)).join();
+ Thread.sleep(TimeUnit.MINUTES.toMillis(1));
+ dorisStandaloneServer.waitingFor(new HostPortWaitStrategy());
+ LOG.info("Doris frontend endpoint and backend endpoint started.");
+ initializeDoris();
Review Comment:
Maybe needn't? i also add `HostPortWaitStrategy` to make sure docker container started.
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] hk-lrzy commented on a diff in pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
hk-lrzy commented on code in PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#discussion_r970600256
##########
seatunnel-e2e/seatunnel-spark-connector-v2-e2e/connector-doris-spark-e2e/src/test/resources/doris/fakesource_to_doris.conf:
##########
@@ -0,0 +1,71 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+######
+###### This config file is a demonstration of streaming processing in seatunnel config
+######
+
+env {
Review Comment:
I know this one. but maybe in spark e2e test, we need'n it, cause by e2e testing will add it automatically
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] hk-lrzy commented on a diff in pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
hk-lrzy commented on code in PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#discussion_r970611081
##########
seatunnel-connectors-v2/connector-doris/pom.xml:
##########
@@ -0,0 +1,95 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You under the Apache License, Version 2.0
+ (the "License"); you may not use this file except in compliance with
+ the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <parent>
+ <artifactId>seatunnel-connectors-v2</artifactId>
+ <groupId>org.apache.seatunnel</groupId>
+ <version>${revision}</version>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <artifactId>connector-doris</artifactId>
+ <properties>
+ <httpclient.version>4.5.13</httpclient.version>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>connector-common</artifactId>
+ <version>${revision}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>seatunnel-format-json</artifactId>
+ <version>${revision}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.httpcomponents</groupId>
+ <artifactId>httpclient</artifactId>
+ <version>${httpclient.version}</version>
+ </dependency>
+ </dependencies>
+
+ <build>
Review Comment:
> client
right. and i already put the pic to show the version difference.
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] hailin0 commented on a diff in pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
hailin0 commented on code in PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#discussion_r964851573
##########
seatunnel-examples/seatunnel-flink-connector-v2-example/pom.xml:
##########
@@ -32,6 +32,16 @@
<flink.scope>compile</flink.scope>
</properties>
+ <dependencyManagement>
+ <dependencies>
+ <dependency>
+ <groupId>commons-io</groupId>
+ <artifactId>commons-io</artifactId>
+ <version>2.8.0</version>
+ </dependency>
+ </dependencies>
+ </dependencyManagement>
+
Review Comment:
revert？
##########
pom.xml:
##########
@@ -277,6 +278,20 @@
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter</artifactId>
+ <version>${junit.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter-api</artifactId>
+ <version>${junit.version}</version>
+ <scope>test</scope>
+ </dependency>
+
Review Comment:
remove.
waiting for dev branch fix
##########
seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/common/HttpClient.java:
##########
@@ -0,0 +1,153 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.doris.common;
+
+import com.google.common.io.Closeables;
+import org.apache.http.HttpEntity;
+import org.apache.http.HttpResponse;
+import org.apache.http.HttpStatus;
+import org.apache.http.client.config.RequestConfig;
+import org.apache.http.client.methods.CloseableHttpResponse;
+import org.apache.http.client.methods.HttpDelete;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.methods.HttpHead;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.client.methods.HttpPut;
+import org.apache.http.client.methods.HttpUriRequest;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.DefaultHttpRequestRetryHandler;
+import org.apache.http.impl.client.DefaultRedirectStrategy;
+import org.apache.http.impl.client.HttpClients;
+import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
+import org.apache.http.util.EntityUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+public class HttpClient {
+ private static final Logger LOG = LoggerFactory.getLogger(HttpClient.class);
+
+ private CloseableHttpClient client;
+ private PoolingHttpClientConnectionManager connectionManager;
+
+ public HttpClient(DorisOptions dorisOptions) {
Review Comment:
reference
https://github.com/apache/incubator-seatunnel/pull/2536#discussion_r956830723
##########
seatunnel-e2e/seatunnel-spark-connector-v2-e2e/connector-doris-spark-e2e/src/test/java/org/apache/seatunnel/e2e/spark/v2/doris/FakeSourceToDorisIT.java:
##########
@@ -0,0 +1,133 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.e2e.spark.v2.doris;
+
+import org.apache.seatunnel.e2e.spark.SparkContainer;
+
+import org.apache.commons.compress.utils.Lists;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.Container;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.containers.wait.strategy.HostPortWaitStrategy;
+import org.testcontainers.lifecycle.Startables;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
+
+public class FakeSourceToDorisIT extends SparkContainer {
+ private static final Logger LOG = LoggerFactory.getLogger(FakeSourceToDorisIT.class);
+
+ private static final String DORIS_DRIVER = "com.mysql.cj.jdbc.Driver";
+ private static final String DORIS_CONNECTION_URL = "jdbc:mysql://localhost:9030?rewriteBatchedStatements=true";
+ private static final String DORIS_PASSWD = "";
+ private static final String DORIS_USERNAME = "root";
+
+ private static final String DORIS_DATABASE = "test";
+ private static final String DORIS_TABLE = "seatunnel";
+ private static final String DORIS_DATABASE_DDL = "CREATE DATABASE IF NOT EXISTS `" + DORIS_DATABASE + "`";
+ private static final String DORIS_USE_DATABASE = "USE `" + DORIS_DATABASE + "`";
+ private static final String DORIS_TABLE_DDL = "CREATE TABLE " +
+ "IF NOT EXISTS `" + DORIS_TABLE + "` " +
+ "(`name` varchar(255) NULL ) " +
+ "ENGINE=OLAP AGGREGATE KEY(`name`) " +
+ "DISTRIBUTED BY HASH(`name`) " +
+ "BUCKETS 1 " +
+ "PROPERTIES " +
+ "( 'replication_allocation' = 'tag.location.default: 1', 'in_memory' = 'false');";
+
+ private static final String DORIS_TABLE_TRUNCATE_TABLE = "TRUNCATE TABLE `" + DORIS_TABLE + "`";
+
+ //thanks zhaomin1432 provided the doris images.
+ private static final String DORIS_IMAGE_NAME = "zhaomin1423/doris:1.0.0-b2";
+ private static final int DORIS_FE_PORT = 8030;
+ private static final int DORIS_QUERY_PORT = 8040;
+ private static final int DORIS_BE_PORT = 9030;
+
+ private GenericContainer<?> dorisStandaloneServer;
+ private Connection connection;
+
+ @BeforeEach
+ public void before() throws InterruptedException {
+ super.before();
+ dorisStandaloneServer = new GenericContainer<>(DORIS_IMAGE_NAME)
+ .withNetwork(NETWORK)
+ .withNetworkAliases("seatunnel-doris-network")
+ .withLogConsumer(new Slf4jLogConsumer(LOG));
+ List<String> portBindings = Lists.newArrayList();
+ portBindings.add(String.format("%s:%s", DORIS_FE_PORT, DORIS_FE_PORT));
+ portBindings.add(String.format("%s:%s", DORIS_QUERY_PORT, DORIS_QUERY_PORT));
+ portBindings.add(String.format("%s:%s", DORIS_BE_PORT, DORIS_BE_PORT));
+ dorisStandaloneServer.setPortBindings(portBindings);
+ Startables.deepStart(Stream.of(dorisStandaloneServer)).join();
+ Thread.sleep(TimeUnit.MINUTES.toMillis(1));
+ dorisStandaloneServer.waitingFor(new HostPortWaitStrategy());
+ LOG.info("Doris frontend endpoint and backend endpoint started.");
+ initializeDoris();
+ }
+
+ private void initializeDoris() {
+ try {
+ connection = getDorisConnection();
+ Statement statement = connection.createStatement();
+ statement.execute(DORIS_DATABASE_DDL);
+ statement.execute(DORIS_USE_DATABASE);
+ statement.execute(DORIS_TABLE_DDL);
+ statement.execute(DORIS_TABLE_TRUNCATE_TABLE);
+ statement.close();
+ } catch (Exception e) {
+ throw new IllegalStateException(e);
+ }
+ }
+
+ private static Connection getDorisConnection() throws ClassNotFoundException, SQLException {
+ Class.forName(DORIS_DRIVER);
+ return DriverManager.getConnection(DORIS_CONNECTION_URL, DORIS_USERNAME, DORIS_PASSWD);
+ }
+
+ @AfterEach
+ public void after() throws SQLException {
+ if (Objects.nonNull(connection)) {
+ connection.close();
+ }
+ if (Objects.nonNull(dorisStandaloneServer)) {
+ dorisStandaloneServer.close();
+ }
+ super.close();
+ }
+
+ //Caused by some reasons, doris image can't run in Mac M1.
+ @Test
+ public void testFakeSourceToConsoleSink() throws IOException, InterruptedException {
+ Container.ExecResult execResult = executeSeaTunnelSparkJob("/doris/fakesource_to_doris.conf");
+ Assertions.assertEquals(0, execResult.getExitCode());
Review Comment:
Read doris data to validate?
##########
seatunnel-examples/seatunnel-flink-connector-v2-example/pom.xml:
##########
@@ -76,6 +86,11 @@
<artifactId>connector-dingtalk</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>connector-datahub</artifactId>
+ <version>${project.version}</version>
+ </dependency>
Review Comment:
revert？
##########
seatunnel-connectors-v2/connector-doris/pom.xml:
##########
@@ -0,0 +1,52 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You under the Apache License, Version 2.0
+ (the "License"); you may not use this file except in compliance with
+ the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <parent>
+ <artifactId>seatunnel-connectors-v2</artifactId>
+ <groupId>org.apache.seatunnel</groupId>
+ <version>${revision}</version>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <artifactId>connector-doris</artifactId>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>connector-common</artifactId>
+ <version>${revision}</version>
Review Comment:
add <scope>provided</scope>
##########
seatunnel-e2e/seatunnel-spark-connector-v2-e2e/connector-doris-spark-e2e/src/test/resources/doris/fakesource_to_doris.conf:
##########
@@ -0,0 +1,64 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+######
+###### This config file is a demonstration of streaming processing in seatunnel config
+######
+
+env {
+ # You can set flink configuration here
+ execution.parallelism = 1
+ job.mode = "BATCH"
+ #execution.checkpoint.interval = 10000
+ #execution.checkpoint.data-uri = "hdfs://localhost:9000/checkpoint"
+}
+
+source {
+ # This is a example source plugin **only for test and demonstrate the feature source plugin**
+ FakeSource {
+ result_table_name = "fake"
+ schema = {
+ fields {
+ name = "string"
Review Comment:
test all datatypes?
##########
seatunnel-e2e/seatunnel-spark-connector-v2-e2e/connector-doris-spark-e2e/src/test/java/org/apache/seatunnel/e2e/spark/v2/doris/FakeSourceToDorisIT.java:
##########
@@ -0,0 +1,133 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.e2e.spark.v2.doris;
+
+import org.apache.seatunnel.e2e.spark.SparkContainer;
+
+import org.apache.commons.compress.utils.Lists;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.Container;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.containers.wait.strategy.HostPortWaitStrategy;
+import org.testcontainers.lifecycle.Startables;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
+
+public class FakeSourceToDorisIT extends SparkContainer {
+ private static final Logger LOG = LoggerFactory.getLogger(FakeSourceToDorisIT.class);
+
+ private static final String DORIS_DRIVER = "com.mysql.cj.jdbc.Driver";
+ private static final String DORIS_CONNECTION_URL = "jdbc:mysql://localhost:9030?rewriteBatchedStatements=true";
+ private static final String DORIS_PASSWD = "";
+ private static final String DORIS_USERNAME = "root";
+
+ private static final String DORIS_DATABASE = "test";
+ private static final String DORIS_TABLE = "seatunnel";
+ private static final String DORIS_DATABASE_DDL = "CREATE DATABASE IF NOT EXISTS `" + DORIS_DATABASE + "`";
+ private static final String DORIS_USE_DATABASE = "USE `" + DORIS_DATABASE + "`";
+ private static final String DORIS_TABLE_DDL = "CREATE TABLE " +
+ "IF NOT EXISTS `" + DORIS_TABLE + "` " +
+ "(`name` varchar(255) NULL ) " +
+ "ENGINE=OLAP AGGREGATE KEY(`name`) " +
+ "DISTRIBUTED BY HASH(`name`) " +
+ "BUCKETS 1 " +
+ "PROPERTIES " +
+ "( 'replication_allocation' = 'tag.location.default: 1', 'in_memory' = 'false');";
+
+ private static final String DORIS_TABLE_TRUNCATE_TABLE = "TRUNCATE TABLE `" + DORIS_TABLE + "`";
+
+ //thanks zhaomin1432 provided the doris images.
+ private static final String DORIS_IMAGE_NAME = "zhaomin1423/doris:1.0.0-b2";
+ private static final int DORIS_FE_PORT = 8030;
+ private static final int DORIS_QUERY_PORT = 8040;
+ private static final int DORIS_BE_PORT = 9030;
+
+ private GenericContainer<?> dorisStandaloneServer;
+ private Connection connection;
+
+ @BeforeEach
+ public void before() throws InterruptedException {
+ super.before();
+ dorisStandaloneServer = new GenericContainer<>(DORIS_IMAGE_NAME)
+ .withNetwork(NETWORK)
+ .withNetworkAliases("seatunnel-doris-network")
+ .withLogConsumer(new Slf4jLogConsumer(LOG));
+ List<String> portBindings = Lists.newArrayList();
+ portBindings.add(String.format("%s:%s", DORIS_FE_PORT, DORIS_FE_PORT));
+ portBindings.add(String.format("%s:%s", DORIS_QUERY_PORT, DORIS_QUERY_PORT));
+ portBindings.add(String.format("%s:%s", DORIS_BE_PORT, DORIS_BE_PORT));
+ dorisStandaloneServer.setPortBindings(portBindings);
+ Startables.deepStart(Stream.of(dorisStandaloneServer)).join();
+ Thread.sleep(TimeUnit.MINUTES.toMillis(1));
+ dorisStandaloneServer.waitingFor(new HostPortWaitStrategy());
+ LOG.info("Doris frontend endpoint and backend endpoint started.");
+ initializeDoris();
+ }
+
+ private void initializeDoris() {
+ try {
+ connection = getDorisConnection();
+ Statement statement = connection.createStatement();
+ statement.execute(DORIS_DATABASE_DDL);
+ statement.execute(DORIS_USE_DATABASE);
+ statement.execute(DORIS_TABLE_DDL);
+ statement.execute(DORIS_TABLE_TRUNCATE_TABLE);
+ statement.close();
+ } catch (Exception e) {
+ throw new IllegalStateException(e);
+ }
+ }
+
+ private static Connection getDorisConnection() throws ClassNotFoundException, SQLException {
+ Class.forName(DORIS_DRIVER);
+ return DriverManager.getConnection(DORIS_CONNECTION_URL, DORIS_USERNAME, DORIS_PASSWD);
+ }
+
+ @AfterEach
+ public void after() throws SQLException {
+ if (Objects.nonNull(connection)) {
+ connection.close();
+ }
+ if (Objects.nonNull(dorisStandaloneServer)) {
+ dorisStandaloneServer.close();
+ }
+ super.close();
+ }
+
+ //Caused by some reasons, doris image can't run in Mac M1.
+ @Test
+ public void testFakeSourceToConsoleSink() throws IOException, InterruptedException {
Review Comment:
rename to `testFakeSourceToDorisSink`
##########
seatunnel-e2e/seatunnel-spark-connector-v2-e2e/connector-spark-e2e-base/src/test/java/org/apache/seatunnel/e2e/spark/SparkContainer.java:
##########
@@ -63,7 +63,7 @@ public abstract class SparkContainer {
private static final int WAIT_SPARK_JOB_SUBMIT = 5000;
@BeforeEach
- public void before() {
+ public void before() throws InterruptedException {
Review Comment:
revert？
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] TyrantLucifer commented on a diff in pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
TyrantLucifer commented on code in PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#discussion_r970351120
##########
seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/sink/DorisWriter.java:
##########
@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.doris.sink;
+
+import org.apache.seatunnel.api.sink.SinkWriter;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import org.apache.seatunnel.common.utils.JsonUtils;
+import org.apache.seatunnel.connectors.doris.common.DorisConstants;
+import org.apache.seatunnel.connectors.doris.common.DorisOptions;
+import org.apache.seatunnel.connectors.doris.sink.loader.DorisStreamLoader;
+import org.apache.seatunnel.connectors.seatunnel.common.sink.AbstractSinkWriter;
+import org.apache.seatunnel.format.json.JsonSerializationSchema;
+
+import org.apache.seatunnel.shade.com.typesafe.config.Config;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import org.apache.commons.collections4.CollectionUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.time.Instant;
+import java.time.ZoneId;
+import java.time.format.DateTimeFormatter;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+
+public class DorisWriter extends AbstractSinkWriter<SeaTunnelRow, Void> {
+ private static final Logger LOG = LoggerFactory.getLogger(DorisWriter.class);
+
+ private static final String LABEL_TEMPLATE = "setunnel_sink_subtask_%s_%s_%s";
+
+ private final SeaTunnelRowType seaTunnelRowType;
+ private final DorisOptions options;
+ private final JsonSerializationSchema serializationSchema;
+ private final DorisLoader<String> loader;
+
+ private final List<JsonNode> batch;
+ private final DateTimeFormatter formatter;
+ private final SinkWriter.Context context;
+
+ public DorisWriter(Config dorisSinkConf,
+ SeaTunnelRowType seaTunnelRowType,
+ SinkWriter.Context context) {
+ this.seaTunnelRowType = seaTunnelRowType;
+ this.options = DorisOptions.fromPluginConfig(dorisSinkConf);
+ this.serializationSchema = new JsonSerializationSchema(seaTunnelRowType);
+ //now we only support stream load, maybe future broker load will implement in seatunnel.
+ this.loader = new DorisStreamLoader(options);
+ this.batch = new ArrayList<>();
+ this.formatter = DateTimeFormatter.ofPattern(DorisConstants.DORIS_LABEL_PATTERN_VALUE)
+ .withZone(ZoneId.systemDefault());
Review Comment:
How about making the time zone a configuration item? Some foreign customers may have a foreign time server, but acutally need to save China time in the database.
##########
seatunnel-core/seatunnel-flink-starter/src/main/java/org/apache/seatunnel/core/starter/flink/execution/SinkExecuteProcessor.java:
##########
@@ -75,7 +75,8 @@ public List<DataStream<Row>> execute(List<DataStream<Row>> upstreamDataStreams)
SeaTunnelSink<SeaTunnelRow, Serializable, Serializable, Serializable> seaTunnelSink = plugins.get(i);
DataStream<Row> stream = fromSourceTable(sinkConfig).orElse(input);
seaTunnelSink.setTypeInfo((SeaTunnelRowType) TypeConverterUtils.convert(stream.getType()));
- stream.sinkTo(new FlinkSink<>(seaTunnelSink));
+ stream.sinkTo(new FlinkSink<>(seaTunnelSink)).name(seaTunnelSink.getPluginName())
Review Comment:
The code is a good optimization point for the whole flink process, but I don't think it should appear in this pr, each pr should have a single responsibility, you should propose another pr to include these changes in it.
##########
seatunnel-connectors-v2/connector-doris/pom.xml:
##########
@@ -0,0 +1,95 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You under the Apache License, Version 2.0
+ (the "License"); you may not use this file except in compliance with
+ the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <parent>
+ <artifactId>seatunnel-connectors-v2</artifactId>
+ <groupId>org.apache.seatunnel</groupId>
+ <version>${revision}</version>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <artifactId>connector-doris</artifactId>
+ <properties>
+ <httpclient.version>4.5.13</httpclient.version>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>connector-common</artifactId>
+ <version>${revision}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>seatunnel-format-json</artifactId>
+ <version>${revision}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.httpcomponents</groupId>
+ <artifactId>httpclient</artifactId>
+ <version>${httpclient.version}</version>
+ </dependency>
+ </dependencies>
+
+ <build>
Review Comment:
I see you had configured the shade plugin, but it is not actually used in the code, what is the point of doing so?
##########
seatunnel-e2e/seatunnel-spark-connector-v2-e2e/connector-doris-spark-e2e/src/test/resources/doris/fakesource_to_doris.conf:
##########
@@ -0,0 +1,71 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+######
+###### This config file is a demonstration of streaming processing in seatunnel config
+######
+
+env {
Review Comment:
Please change the env parameter to spark related parameters.
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] hk-lrzy commented on a diff in pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
hk-lrzy commented on code in PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#discussion_r970584745
##########
seatunnel-connectors-v2/connector-doris/pom.xml:
##########
@@ -0,0 +1,95 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You under the Apache License, Version 2.0
+ (the "License"); you may not use this file except in compliance with
+ the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <parent>
+ <artifactId>seatunnel-connectors-v2</artifactId>
+ <groupId>org.apache.seatunnel</groupId>
+ <version>${revision}</version>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <artifactId>connector-doris</artifactId>
+ <properties>
+ <httpclient.version>4.5.13</httpclient.version>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>connector-common</artifactId>
+ <version>${revision}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>seatunnel-format-json</artifactId>
+ <version>${revision}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.httpcomponents</groupId>
+ <artifactId>httpclient</artifactId>
+ <version>${httpclient.version}</version>
+ </dependency>
+ </dependencies>
+
+ <build>
Review Comment:
Becasue it conflict with spark. you can run the spark e2e testing without the shaded plugin, it will have exception.
And i see the dependency in the spark's lib, the version is `4.5.6`, so we need use shaded to avoid it.
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] hk-lrzy commented on a diff in pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
hk-lrzy commented on code in PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#discussion_r970611081
##########
seatunnel-connectors-v2/connector-doris/pom.xml:
##########
@@ -0,0 +1,95 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You under the Apache License, Version 2.0
+ (the "License"); you may not use this file except in compliance with
+ the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <parent>
+ <artifactId>seatunnel-connectors-v2</artifactId>
+ <groupId>org.apache.seatunnel</groupId>
+ <version>${revision}</version>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <artifactId>connector-doris</artifactId>
+ <properties>
+ <httpclient.version>4.5.13</httpclient.version>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>connector-common</artifactId>
+ <version>${revision}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>seatunnel-format-json</artifactId>
+ <version>${revision}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.httpcomponents</groupId>
+ <artifactId>httpclient</artifactId>
+ <version>${httpclient.version}</version>
+ </dependency>
+ </dependencies>
+
+ <build>
Review Comment:
> client
right. and i put the screenshot to show the version difference.
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] hailin0 commented on a diff in pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
hailin0 commented on code in PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#discussion_r970742612
##########
seatunnel-e2e/seatunnel-spark-connector-v2-e2e/connector-doris-spark-e2e/src/test/java/org/apache/seatunnel/e2e/spark/v2/doris/FakeSourceToDorisIT.java:
##########
@@ -0,0 +1,150 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.e2e.spark.v2.doris;
+
+import org.apache.seatunnel.e2e.spark.SparkContainer;
+
+import org.apache.commons.compress.utils.Lists;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.Container;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.containers.wait.strategy.HostPortWaitStrategy;
+import org.testcontainers.lifecycle.Startables;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
+
+public class FakeSourceToDorisIT extends SparkContainer {
+ private static final Logger LOG = LoggerFactory.getLogger(FakeSourceToDorisIT.class);
+
+ private static final String DORIS_DRIVER = "com.mysql.cj.jdbc.Driver";
+ private static final String DORIS_CONNECTION_URL = "jdbc:mysql://localhost:9030?rewriteBatchedStatements=true";
+ private static final String DORIS_PASSWD = "";
+ private static final String DORIS_USERNAME = "root";
+
+ private static final String DORIS_DATABASE = "test";
+ private static final String DORIS_TABLE = "seatunnel";
+ private static final String DORIS_DATABASE_DDL = "CREATE DATABASE IF NOT EXISTS `" + DORIS_DATABASE + "`";
+ private static final String DORIS_USE_DATABASE = "USE `" + DORIS_DATABASE + "`";
+ private static final String DORIS_TABLE_DDL = "CREATE TABLE IF NOT EXISTS `" + DORIS_DATABASE + "`.`" + DORIS_TABLE + "` ( " +
+ " `user_id` LARGEINT NOT NULL COMMENT 'id'," +
+ " `city` VARCHAR(20) COMMENT 'city'," +
+ " `age` SMALLINT COMMENT 'age'," +
+ " `sex` TINYINT COMMENT 'sec'," +
+ " `cost` BIGINT SUM DEFAULT '0' ," +
+ " `max_dwell_time` INT MAX DEFAULT '0' ," +
+ " `min_dwell_time` INT MIN DEFAULT '99999'" +
+ ") AGGREGATE KEY(`user_id`, `city`, `age`, `sex`) DISTRIBUTED BY HASH(`user_id`) BUCKETS 1 PROPERTIES (" +
+ " 'replication_allocation' = 'tag.location.default: 1'" +
+ ");";
+
+ private static final String DORIS_TABLE_TRUNCATE_TABLE = "TRUNCATE TABLE `" + DORIS_TABLE + "`";
+ private static final String DORIS_SELECT_TABLE = "SELECT COUNT(*) FROM `" + DORIS_TABLE + "`";
+
+ //thanks zhaomin1432 provided the doris images.
+ private static final String DORIS_IMAGE_NAME = "zhaomin1423/doris:1.0.0-b2";
+ private static final int DORIS_FE_PORT = 8030;
+ private static final int DORIS_QUERY_PORT = 8040;
+ private static final int DORIS_BE_PORT = 9030;
+
+ private GenericContainer<?> dorisStandaloneServer;
+ private Connection connection;
+
+ @BeforeEach
+ public void beforeInDoris() throws InterruptedException {
+ super.before();
+ dorisStandaloneServer = new GenericContainer<>(DORIS_IMAGE_NAME)
+ .withNetwork(NETWORK)
+ .withNetworkAliases("seatunnel-doris-network")
+ .withLogConsumer(new Slf4jLogConsumer(LOG));
+ List<String> portBindings = Lists.newArrayList();
+ portBindings.add(String.format("%s:%s", DORIS_FE_PORT, DORIS_FE_PORT));
+ portBindings.add(String.format("%s:%s", DORIS_QUERY_PORT, DORIS_QUERY_PORT));
Review Comment:
as above
##########
seatunnel-e2e/seatunnel-spark-connector-v2-e2e/connector-doris-spark-e2e/src/test/java/org/apache/seatunnel/e2e/spark/v2/doris/FakeSourceToDorisIT.java:
##########
@@ -0,0 +1,150 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.e2e.spark.v2.doris;
+
+import org.apache.seatunnel.e2e.spark.SparkContainer;
+
+import org.apache.commons.compress.utils.Lists;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.Container;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.containers.wait.strategy.HostPortWaitStrategy;
+import org.testcontainers.lifecycle.Startables;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
+
+public class FakeSourceToDorisIT extends SparkContainer {
+ private static final Logger LOG = LoggerFactory.getLogger(FakeSourceToDorisIT.class);
+
+ private static final String DORIS_DRIVER = "com.mysql.cj.jdbc.Driver";
+ private static final String DORIS_CONNECTION_URL = "jdbc:mysql://localhost:9030?rewriteBatchedStatements=true";
+ private static final String DORIS_PASSWD = "";
+ private static final String DORIS_USERNAME = "root";
+
+ private static final String DORIS_DATABASE = "test";
+ private static final String DORIS_TABLE = "seatunnel";
+ private static final String DORIS_DATABASE_DDL = "CREATE DATABASE IF NOT EXISTS `" + DORIS_DATABASE + "`";
+ private static final String DORIS_USE_DATABASE = "USE `" + DORIS_DATABASE + "`";
+ private static final String DORIS_TABLE_DDL = "CREATE TABLE IF NOT EXISTS `" + DORIS_DATABASE + "`.`" + DORIS_TABLE + "` ( " +
+ " `user_id` LARGEINT NOT NULL COMMENT 'id'," +
+ " `city` VARCHAR(20) COMMENT 'city'," +
+ " `age` SMALLINT COMMENT 'age'," +
+ " `sex` TINYINT COMMENT 'sec'," +
+ " `cost` BIGINT SUM DEFAULT '0' ," +
+ " `max_dwell_time` INT MAX DEFAULT '0' ," +
+ " `min_dwell_time` INT MIN DEFAULT '99999'" +
+ ") AGGREGATE KEY(`user_id`, `city`, `age`, `sex`) DISTRIBUTED BY HASH(`user_id`) BUCKETS 1 PROPERTIES (" +
+ " 'replication_allocation' = 'tag.location.default: 1'" +
+ ");";
+
+ private static final String DORIS_TABLE_TRUNCATE_TABLE = "TRUNCATE TABLE `" + DORIS_TABLE + "`";
+ private static final String DORIS_SELECT_TABLE = "SELECT COUNT(*) FROM `" + DORIS_TABLE + "`";
+
+ //thanks zhaomin1432 provided the doris images.
+ private static final String DORIS_IMAGE_NAME = "zhaomin1423/doris:1.0.0-b2";
+ private static final int DORIS_FE_PORT = 8030;
+ private static final int DORIS_QUERY_PORT = 8040;
+ private static final int DORIS_BE_PORT = 9030;
+
+ private GenericContainer<?> dorisStandaloneServer;
+ private Connection connection;
+
+ @BeforeEach
+ public void beforeInDoris() throws InterruptedException {
+ super.before();
+ dorisStandaloneServer = new GenericContainer<>(DORIS_IMAGE_NAME)
+ .withNetwork(NETWORK)
+ .withNetworkAliases("seatunnel-doris-network")
+ .withLogConsumer(new Slf4jLogConsumer(LOG));
+ List<String> portBindings = Lists.newArrayList();
+ portBindings.add(String.format("%s:%s", DORIS_FE_PORT, DORIS_FE_PORT));
+ portBindings.add(String.format("%s:%s", DORIS_QUERY_PORT, DORIS_QUERY_PORT));
+ portBindings.add(String.format("%s:%s", DORIS_BE_PORT, DORIS_BE_PORT));
+ dorisStandaloneServer.setPortBindings(portBindings);
+ Startables.deepStart(Stream.of(dorisStandaloneServer)).join();
+ Thread.sleep(TimeUnit.MINUTES.toMillis(1));
+ dorisStandaloneServer.waitingFor(new HostPortWaitStrategy());
+ LOG.info("Doris frontend endpoint and backend endpoint started.");
+ initializeDoris();
Review Comment:
as above
##########
seatunnel-e2e/seatunnel-flink-connector-v2-e2e/connector-flink-e2e-base/src/test/resources/log4j.properties:
##########
@@ -0,0 +1,22 @@
+#
Review Comment:
move this file to `connector-doris-flink-e2e` module?
reference
https://github.com/apache/incubator-seatunnel/tree/dev/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/connector-jdbc-flink-e2e/src/test/resources
##########
seatunnel-e2e/seatunnel-flink-connector-v2-e2e/connector-doris-flink-e2e/src/test/java/org/apache/seatunnel/e2e/flink/v2/doris/FakeSourceToDorisIT.java:
##########
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.e2e.flink.v2.doris;
+
+import org.apache.seatunnel.e2e.flink.FlinkContainer;
+
+import com.google.common.collect.Lists;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.Container;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.containers.wait.strategy.HostPortWaitStrategy;
+import org.testcontainers.lifecycle.Startables;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
+
+public class FakeSourceToDorisIT extends FlinkContainer {
+ private static final Logger LOG = LoggerFactory.getLogger(FakeSourceToDorisIT.class);
+
+ private static final String DORIS_DRIVER = "com.mysql.cj.jdbc.Driver";
+ private static final String DORIS_CONNECTION_URL = "jdbc:mysql://localhost:9030?rewriteBatchedStatements=true";
+ private static final String DORIS_PASSWD = "";
+ private static final String DORIS_USERNAME = "root";
+
+ private static final String DORIS_DATABASE = "test";
+ private static final String DORIS_TABLE = "seatunnel";
+ private static final String DORIS_DATABASE_DDL = "CREATE DATABASE IF NOT EXISTS `" + DORIS_DATABASE + "`";
+ private static final String DORIS_USE_DATABASE = "USE `" + DORIS_DATABASE + "`";
+ private static final String DORIS_TABLE_DDL = "CREATE TABLE IF NOT EXISTS `" + DORIS_DATABASE + "`.`" + DORIS_TABLE + "` ( " +
+ " `user_id` LARGEINT NOT NULL COMMENT 'id'," +
+ " `date` DATE NOT NULL COMMENT 'date'," +
+ " `city` VARCHAR(20) COMMENT 'city'," +
+ " `age` SMALLINT COMMENT 'age'," +
+ " `sex` TINYINT COMMENT 'sec'," +
+ " `last_visit_date` DATETIME REPLACE DEFAULT '1970-01-01 00:00:00' ," +
+ " `cost` BIGINT SUM DEFAULT '0' ," +
+ " `max_dwell_time` INT MAX DEFAULT '0' ," +
+ " `min_dwell_time` INT MIN DEFAULT '99999'" +
+ ") AGGREGATE KEY(`user_id`, `date`, `city`, `age`, `sex`) DISTRIBUTED BY HASH(`user_id`) BUCKETS 1 PROPERTIES (" +
+ " 'replication_allocation' = 'tag.location.default: 1'" +
+ ");";
+
+ private static final String DORIS_TRUNCATE_TABLE = "TRUNCATE TABLE `" + DORIS_TABLE + "`";
+ private static final String DORIS_SELECT_TABLE = "SELECT COUNT(*) FROM `" + DORIS_TABLE + "`";
+
+ //thanks zhaomin1432 provided the doris images.
+ private static final String DORIS_IMAGE_NAME = "zhaomin1423/doris:1.0.0-b2";
+ private static final int DORIS_FE_PORT = 8030;
+ private static final int DORIS_QUERY_PORT = 8040;
+ private static final int DORIS_BE_PORT = 9030;
+
+ private GenericContainer<?> dorisStandaloneServer;
+ private Connection connection;
+
+ @BeforeEach
+ public void beforeEach() throws InterruptedException {
+ dorisStandaloneServer = new GenericContainer<>(DORIS_IMAGE_NAME)
+ .withNetwork(NETWORK)
+ .withNetworkAliases("seatunnel-doris-network")
+ .withLogConsumer(new Slf4jLogConsumer(LOG));
+ List<String> portBindings = Lists.newArrayList();
+ portBindings.add(String.format("%s:%s", DORIS_FE_PORT, DORIS_FE_PORT));
+ portBindings.add(String.format("%s:%s", DORIS_QUERY_PORT, DORIS_QUERY_PORT));
+ portBindings.add(String.format("%s:%s", DORIS_BE_PORT, DORIS_BE_PORT));
+ dorisStandaloneServer.setPortBindings(portBindings);
+ Startables.deepStart(Stream.of(dorisStandaloneServer)).join();
+ Thread.sleep(TimeUnit.MINUTES.toMillis(1));
+ dorisStandaloneServer.waitingFor(new HostPortWaitStrategy());
+ LOG.info("Doris frontend endpoint and backend endpoint started.");
+ initializeDoris();
Review Comment:
Use awaitility framework？
reference #2677
##########
seatunnel-e2e/seatunnel-flink-connector-v2-e2e/connector-doris-flink-e2e/src/test/java/org/apache/seatunnel/e2e/flink/v2/doris/FakeSourceToDorisIT.java:
##########
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.e2e.flink.v2.doris;
+
+import org.apache.seatunnel.e2e.flink.FlinkContainer;
+
+import com.google.common.collect.Lists;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.Container;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.containers.wait.strategy.HostPortWaitStrategy;
+import org.testcontainers.lifecycle.Startables;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
+
+public class FakeSourceToDorisIT extends FlinkContainer {
+ private static final Logger LOG = LoggerFactory.getLogger(FakeSourceToDorisIT.class);
+
+ private static final String DORIS_DRIVER = "com.mysql.cj.jdbc.Driver";
+ private static final String DORIS_CONNECTION_URL = "jdbc:mysql://localhost:9030?rewriteBatchedStatements=true";
+ private static final String DORIS_PASSWD = "";
+ private static final String DORIS_USERNAME = "root";
+
+ private static final String DORIS_DATABASE = "test";
+ private static final String DORIS_TABLE = "seatunnel";
+ private static final String DORIS_DATABASE_DDL = "CREATE DATABASE IF NOT EXISTS `" + DORIS_DATABASE + "`";
+ private static final String DORIS_USE_DATABASE = "USE `" + DORIS_DATABASE + "`";
+ private static final String DORIS_TABLE_DDL = "CREATE TABLE IF NOT EXISTS `" + DORIS_DATABASE + "`.`" + DORIS_TABLE + "` ( " +
+ " `user_id` LARGEINT NOT NULL COMMENT 'id'," +
+ " `date` DATE NOT NULL COMMENT 'date'," +
+ " `city` VARCHAR(20) COMMENT 'city'," +
+ " `age` SMALLINT COMMENT 'age'," +
+ " `sex` TINYINT COMMENT 'sec'," +
+ " `last_visit_date` DATETIME REPLACE DEFAULT '1970-01-01 00:00:00' ," +
+ " `cost` BIGINT SUM DEFAULT '0' ," +
+ " `max_dwell_time` INT MAX DEFAULT '0' ," +
+ " `min_dwell_time` INT MIN DEFAULT '99999'" +
+ ") AGGREGATE KEY(`user_id`, `date`, `city`, `age`, `sex`) DISTRIBUTED BY HASH(`user_id`) BUCKETS 1 PROPERTIES (" +
+ " 'replication_allocation' = 'tag.location.default: 1'" +
+ ");";
+
+ private static final String DORIS_TRUNCATE_TABLE = "TRUNCATE TABLE `" + DORIS_TABLE + "`";
+ private static final String DORIS_SELECT_TABLE = "SELECT COUNT(*) FROM `" + DORIS_TABLE + "`";
+
+ //thanks zhaomin1432 provided the doris images.
+ private static final String DORIS_IMAGE_NAME = "zhaomin1423/doris:1.0.0-b2";
+ private static final int DORIS_FE_PORT = 8030;
+ private static final int DORIS_QUERY_PORT = 8040;
+ private static final int DORIS_BE_PORT = 9030;
+
+ private GenericContainer<?> dorisStandaloneServer;
+ private Connection connection;
+
+ @BeforeEach
+ public void beforeEach() throws InterruptedException {
+ dorisStandaloneServer = new GenericContainer<>(DORIS_IMAGE_NAME)
+ .withNetwork(NETWORK)
+ .withNetworkAliases("seatunnel-doris-network")
+ .withLogConsumer(new Slf4jLogConsumer(LOG));
+ List<String> portBindings = Lists.newArrayList();
+ portBindings.add(String.format("%s:%s", DORIS_FE_PORT, DORIS_FE_PORT));
+ portBindings.add(String.format("%s:%s", DORIS_QUERY_PORT, DORIS_QUERY_PORT));
Review Comment:
unused？
##########
seatunnel-e2e/seatunnel-flink-connector-v2-e2e/connector-doris-flink-e2e/src/test/java/org/apache/seatunnel/e2e/flink/v2/doris/FakeSourceToDorisIT.java:
##########
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.e2e.flink.v2.doris;
+
+import org.apache.seatunnel.e2e.flink.FlinkContainer;
+
+import com.google.common.collect.Lists;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.Container;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.containers.wait.strategy.HostPortWaitStrategy;
+import org.testcontainers.lifecycle.Startables;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
+
+public class FakeSourceToDorisIT extends FlinkContainer {
+ private static final Logger LOG = LoggerFactory.getLogger(FakeSourceToDorisIT.class);
+
+ private static final String DORIS_DRIVER = "com.mysql.cj.jdbc.Driver";
+ private static final String DORIS_CONNECTION_URL = "jdbc:mysql://localhost:9030?rewriteBatchedStatements=true";
+ private static final String DORIS_PASSWD = "";
+ private static final String DORIS_USERNAME = "root";
+
+ private static final String DORIS_DATABASE = "test";
+ private static final String DORIS_TABLE = "seatunnel";
+ private static final String DORIS_DATABASE_DDL = "CREATE DATABASE IF NOT EXISTS `" + DORIS_DATABASE + "`";
+ private static final String DORIS_USE_DATABASE = "USE `" + DORIS_DATABASE + "`";
+ private static final String DORIS_TABLE_DDL = "CREATE TABLE IF NOT EXISTS `" + DORIS_DATABASE + "`.`" + DORIS_TABLE + "` ( " +
+ " `user_id` LARGEINT NOT NULL COMMENT 'id'," +
+ " `date` DATE NOT NULL COMMENT 'date'," +
+ " `city` VARCHAR(20) COMMENT 'city'," +
+ " `age` SMALLINT COMMENT 'age'," +
+ " `sex` TINYINT COMMENT 'sec'," +
+ " `last_visit_date` DATETIME REPLACE DEFAULT '1970-01-01 00:00:00' ," +
+ " `cost` BIGINT SUM DEFAULT '0' ," +
+ " `max_dwell_time` INT MAX DEFAULT '0' ," +
+ " `min_dwell_time` INT MIN DEFAULT '99999'" +
+ ") AGGREGATE KEY(`user_id`, `date`, `city`, `age`, `sex`) DISTRIBUTED BY HASH(`user_id`) BUCKETS 1 PROPERTIES (" +
+ " 'replication_allocation' = 'tag.location.default: 1'" +
+ ");";
+
+ private static final String DORIS_TRUNCATE_TABLE = "TRUNCATE TABLE `" + DORIS_TABLE + "`";
+ private static final String DORIS_SELECT_TABLE = "SELECT COUNT(*) FROM `" + DORIS_TABLE + "`";
+
+ //thanks zhaomin1432 provided the doris images.
+ private static final String DORIS_IMAGE_NAME = "zhaomin1423/doris:1.0.0-b2";
Review Comment:
why not use `apache/doris` image?
##########
seatunnel-e2e/seatunnel-flink-connector-v2-e2e/connector-doris-flink-e2e/src/test/resources/doris/fakesource_to_doris.conf:
##########
@@ -0,0 +1,73 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+######
+###### This config file is a demonstration of streaming processing in seatunnel config
+######
+
+env {
+ # You can set flink configuration here
+ execution.parallelism = 1
+ job.mode = "BATCH"
+ #execution.checkpoint.interval = 10000
+ #execution.checkpoint.data-uri = "hdfs://localhost:9000/checkpoint"
+}
+
+source {
+ # This is a example source plugin **only for test and demonstrate the feature source plugin**
+ FakeSource {
+ result_table_name = "fake"
+ schema = {
+ fields {
+ user_id = "bigint",
+ date = "date",
+ city = "string",
+ age = "smallint",
+ sex = "smallint",
+ last_visit_date = "timestamp",
+ cost = "bigint",
+ max_dwell_time = "int",
+ min_dwell_time = "int",
+ price = "double"
Review Comment:
test all datatype?
reference
https://github.com/apache/incubator-seatunnel/blob/dev/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/connector-mongodb-flink-e2e/src/test/resources/mongodb/fake_to_mongodb.conf#L37
##########
seatunnel-e2e/seatunnel-spark-connector-v2-e2e/connector-doris-spark-e2e/src/test/java/org/apache/seatunnel/e2e/spark/v2/doris/FakeSourceToDorisIT.java:
##########
@@ -0,0 +1,150 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.e2e.spark.v2.doris;
+
+import org.apache.seatunnel.e2e.spark.SparkContainer;
+
+import org.apache.commons.compress.utils.Lists;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.Container;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.containers.wait.strategy.HostPortWaitStrategy;
+import org.testcontainers.lifecycle.Startables;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
+
+public class FakeSourceToDorisIT extends SparkContainer {
+ private static final Logger LOG = LoggerFactory.getLogger(FakeSourceToDorisIT.class);
+
+ private static final String DORIS_DRIVER = "com.mysql.cj.jdbc.Driver";
+ private static final String DORIS_CONNECTION_URL = "jdbc:mysql://localhost:9030?rewriteBatchedStatements=true";
+ private static final String DORIS_PASSWD = "";
+ private static final String DORIS_USERNAME = "root";
+
+ private static final String DORIS_DATABASE = "test";
+ private static final String DORIS_TABLE = "seatunnel";
+ private static final String DORIS_DATABASE_DDL = "CREATE DATABASE IF NOT EXISTS `" + DORIS_DATABASE + "`";
+ private static final String DORIS_USE_DATABASE = "USE `" + DORIS_DATABASE + "`";
+ private static final String DORIS_TABLE_DDL = "CREATE TABLE IF NOT EXISTS `" + DORIS_DATABASE + "`.`" + DORIS_TABLE + "` ( " +
+ " `user_id` LARGEINT NOT NULL COMMENT 'id'," +
+ " `city` VARCHAR(20) COMMENT 'city'," +
+ " `age` SMALLINT COMMENT 'age'," +
+ " `sex` TINYINT COMMENT 'sec'," +
+ " `cost` BIGINT SUM DEFAULT '0' ," +
+ " `max_dwell_time` INT MAX DEFAULT '0' ," +
+ " `min_dwell_time` INT MIN DEFAULT '99999'" +
+ ") AGGREGATE KEY(`user_id`, `city`, `age`, `sex`) DISTRIBUTED BY HASH(`user_id`) BUCKETS 1 PROPERTIES (" +
+ " 'replication_allocation' = 'tag.location.default: 1'" +
+ ");";
+
+ private static final String DORIS_TABLE_TRUNCATE_TABLE = "TRUNCATE TABLE `" + DORIS_TABLE + "`";
+ private static final String DORIS_SELECT_TABLE = "SELECT COUNT(*) FROM `" + DORIS_TABLE + "`";
+
+ //thanks zhaomin1432 provided the doris images.
+ private static final String DORIS_IMAGE_NAME = "zhaomin1423/doris:1.0.0-b2";
Review Comment:
as above
##########
seatunnel-e2e/seatunnel-spark-connector-v2-e2e/connector-doris-spark-e2e/src/test/resources/doris/fakesource_to_doris.conf:
##########
@@ -0,0 +1,75 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+######
+###### This config file is a demonstration of streaming processing in seatunnel config
+######
+
+env {
+ # You can set flink configuration here
+ job.mode = "BATCH"
+ spark.app.name = "SeaTunnel"
+ spark.executor.instances = 2
+ spark.executor.cores = 1
+ spark.executor.memory = "1g"
+ spark.master = local
+ #execution.checkpoint.interval = 10000
+ #execution.checkpoint.data-uri = "hdfs://localhost:9000/checkpoint"
+}
+
+source {
+ # This is a example source plugin **only for test and demonstrate the feature source plugin**
+ FakeSource {
+ result_table_name = "fake"
+ schema = {
+ fields {
+ user_id = "bigint",
+ city = "string",
+ age = "smallint",
+ sex = "smallint",
+ cost = "bigint",
+ max_dwell_time = "int",
+ min_dwell_time = "int",
+ price = "double"
Review Comment:
as above
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] hk-lrzy commented on a diff in pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
hk-lrzy commented on code in PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#discussion_r969116151
##########
seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/common/HttpClient.java:
##########
@@ -0,0 +1,153 @@
+/*
Review Comment:
Just like `seatunnel-formats`
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] hk-lrzy commented on pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris.
Posted by GitBox <gi...@apache.org>.
hk-lrzy commented on PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#issuecomment-1233739470
> flink
Sure, i am try to , but there have no avaliable docker image for doris now.
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] hk-lrzy commented on pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
hk-lrzy commented on PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#issuecomment-1239280862
> Need to add spark e2e test
i already fix it, plz take a look again @ic4y
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] hk-lrzy commented on a diff in pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
hk-lrzy commented on code in PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#discussion_r970585997
##########
seatunnel-core/seatunnel-flink-starter/src/main/java/org/apache/seatunnel/core/starter/flink/execution/SinkExecuteProcessor.java:
##########
@@ -75,7 +75,8 @@ public List<DataStream<Row>> execute(List<DataStream<Row>> upstreamDataStreams)
SeaTunnelSink<SeaTunnelRow, Serializable, Serializable, Serializable> seaTunnelSink = plugins.get(i);
DataStream<Row> stream = fromSourceTable(sinkConfig).orElse(input);
seaTunnelSink.setTypeInfo((SeaTunnelRowType) TypeConverterUtils.convert(stream.getType()));
- stream.sinkTo(new FlinkSink<>(seaTunnelSink));
+ stream.sinkTo(new FlinkSink<>(seaTunnelSink)).name(seaTunnelSink.getPluginName())
Review Comment:
Sure, you are right, i just test it and find this issue. i will remove it and submit a new one to fix it.
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] hk-lrzy commented on a diff in pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
hk-lrzy commented on code in PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#discussion_r970613122
##########
seatunnel-e2e/seatunnel-spark-connector-v2-e2e/connector-doris-spark-e2e/src/test/resources/doris/fakesource_to_doris.conf:
##########
@@ -0,0 +1,71 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+######
+###### This config file is a demonstration of streaming processing in seatunnel config
+######
+
+env {
Review Comment:
I push new commit to fixe this comment, plz take a look again thx.
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] hailin0 commented on a diff in pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
hailin0 commented on code in PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#discussion_r970759451
##########
seatunnel-e2e/seatunnel-flink-connector-v2-e2e/connector-doris-flink-e2e/src/test/java/org/apache/seatunnel/e2e/flink/v2/doris/FakeSourceToDorisIT.java:
##########
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.e2e.flink.v2.doris;
+
+import org.apache.seatunnel.e2e.flink.FlinkContainer;
+
+import com.google.common.collect.Lists;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.Container;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.containers.wait.strategy.HostPortWaitStrategy;
+import org.testcontainers.lifecycle.Startables;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
+
+public class FakeSourceToDorisIT extends FlinkContainer {
+ private static final Logger LOG = LoggerFactory.getLogger(FakeSourceToDorisIT.class);
+
+ private static final String DORIS_DRIVER = "com.mysql.cj.jdbc.Driver";
+ private static final String DORIS_CONNECTION_URL = "jdbc:mysql://localhost:9030?rewriteBatchedStatements=true";
+ private static final String DORIS_PASSWD = "";
+ private static final String DORIS_USERNAME = "root";
+
+ private static final String DORIS_DATABASE = "test";
+ private static final String DORIS_TABLE = "seatunnel";
+ private static final String DORIS_DATABASE_DDL = "CREATE DATABASE IF NOT EXISTS `" + DORIS_DATABASE + "`";
+ private static final String DORIS_USE_DATABASE = "USE `" + DORIS_DATABASE + "`";
+ private static final String DORIS_TABLE_DDL = "CREATE TABLE IF NOT EXISTS `" + DORIS_DATABASE + "`.`" + DORIS_TABLE + "` ( " +
+ " `user_id` LARGEINT NOT NULL COMMENT 'id'," +
+ " `date` DATE NOT NULL COMMENT 'date'," +
+ " `city` VARCHAR(20) COMMENT 'city'," +
+ " `age` SMALLINT COMMENT 'age'," +
+ " `sex` TINYINT COMMENT 'sec'," +
+ " `last_visit_date` DATETIME REPLACE DEFAULT '1970-01-01 00:00:00' ," +
+ " `cost` BIGINT SUM DEFAULT '0' ," +
+ " `max_dwell_time` INT MAX DEFAULT '0' ," +
+ " `min_dwell_time` INT MIN DEFAULT '99999'" +
+ ") AGGREGATE KEY(`user_id`, `date`, `city`, `age`, `sex`) DISTRIBUTED BY HASH(`user_id`) BUCKETS 1 PROPERTIES (" +
+ " 'replication_allocation' = 'tag.location.default: 1'" +
+ ");";
+
+ private static final String DORIS_TRUNCATE_TABLE = "TRUNCATE TABLE `" + DORIS_TABLE + "`";
+ private static final String DORIS_SELECT_TABLE = "SELECT COUNT(*) FROM `" + DORIS_TABLE + "`";
+
+ //thanks zhaomin1432 provided the doris images.
+ private static final String DORIS_IMAGE_NAME = "zhaomin1423/doris:1.0.0-b2";
+ private static final int DORIS_FE_PORT = 8030;
+ private static final int DORIS_QUERY_PORT = 8040;
+ private static final int DORIS_BE_PORT = 9030;
+
+ private GenericContainer<?> dorisStandaloneServer;
+ private Connection connection;
+
+ @BeforeEach
+ public void beforeEach() throws InterruptedException {
+ dorisStandaloneServer = new GenericContainer<>(DORIS_IMAGE_NAME)
+ .withNetwork(NETWORK)
+ .withNetworkAliases("seatunnel-doris-network")
+ .withLogConsumer(new Slf4jLogConsumer(LOG));
+ List<String> portBindings = Lists.newArrayList();
+ portBindings.add(String.format("%s:%s", DORIS_FE_PORT, DORIS_FE_PORT));
+ portBindings.add(String.format("%s:%s", DORIS_QUERY_PORT, DORIS_QUERY_PORT));
Review Comment:
DORIS_QUERY_PORT
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] hk-lrzy commented on a diff in pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
hk-lrzy commented on code in PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#discussion_r970886475
##########
seatunnel-e2e/seatunnel-flink-connector-v2-e2e/connector-doris-flink-e2e/src/test/java/org/apache/seatunnel/e2e/flink/v2/doris/FakeSourceToDorisIT.java:
##########
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.e2e.flink.v2.doris;
+
+import org.apache.seatunnel.e2e.flink.FlinkContainer;
+
+import com.google.common.collect.Lists;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.Container;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.containers.wait.strategy.HostPortWaitStrategy;
+import org.testcontainers.lifecycle.Startables;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
+
+public class FakeSourceToDorisIT extends FlinkContainer {
+ private static final Logger LOG = LoggerFactory.getLogger(FakeSourceToDorisIT.class);
+
+ private static final String DORIS_DRIVER = "com.mysql.cj.jdbc.Driver";
+ private static final String DORIS_CONNECTION_URL = "jdbc:mysql://localhost:9030?rewriteBatchedStatements=true";
+ private static final String DORIS_PASSWD = "";
+ private static final String DORIS_USERNAME = "root";
+
+ private static final String DORIS_DATABASE = "test";
+ private static final String DORIS_TABLE = "seatunnel";
+ private static final String DORIS_DATABASE_DDL = "CREATE DATABASE IF NOT EXISTS `" + DORIS_DATABASE + "`";
+ private static final String DORIS_USE_DATABASE = "USE `" + DORIS_DATABASE + "`";
+ private static final String DORIS_TABLE_DDL = "CREATE TABLE IF NOT EXISTS `" + DORIS_DATABASE + "`.`" + DORIS_TABLE + "` ( " +
+ " `user_id` LARGEINT NOT NULL COMMENT 'id'," +
+ " `date` DATE NOT NULL COMMENT 'date'," +
+ " `city` VARCHAR(20) COMMENT 'city'," +
+ " `age` SMALLINT COMMENT 'age'," +
+ " `sex` TINYINT COMMENT 'sec'," +
+ " `last_visit_date` DATETIME REPLACE DEFAULT '1970-01-01 00:00:00' ," +
+ " `cost` BIGINT SUM DEFAULT '0' ," +
+ " `max_dwell_time` INT MAX DEFAULT '0' ," +
+ " `min_dwell_time` INT MIN DEFAULT '99999'" +
+ ") AGGREGATE KEY(`user_id`, `date`, `city`, `age`, `sex`) DISTRIBUTED BY HASH(`user_id`) BUCKETS 1 PROPERTIES (" +
+ " 'replication_allocation' = 'tag.location.default: 1'" +
+ ");";
+
+ private static final String DORIS_TRUNCATE_TABLE = "TRUNCATE TABLE `" + DORIS_TABLE + "`";
+ private static final String DORIS_SELECT_TABLE = "SELECT COUNT(*) FROM `" + DORIS_TABLE + "`";
+
+ //thanks zhaomin1432 provided the doris images.
+ private static final String DORIS_IMAGE_NAME = "zhaomin1423/doris:1.0.0-b2";
+ private static final int DORIS_FE_PORT = 8030;
+ private static final int DORIS_QUERY_PORT = 8040;
+ private static final int DORIS_BE_PORT = 9030;
+
+ private GenericContainer<?> dorisStandaloneServer;
+ private Connection connection;
+
+ @BeforeEach
+ public void beforeEach() throws InterruptedException {
+ dorisStandaloneServer = new GenericContainer<>(DORIS_IMAGE_NAME)
+ .withNetwork(NETWORK)
+ .withNetworkAliases("seatunnel-doris-network")
+ .withLogConsumer(new Slf4jLogConsumer(LOG));
+ List<String> portBindings = Lists.newArrayList();
+ portBindings.add(String.format("%s:%s", DORIS_FE_PORT, DORIS_FE_PORT));
+ portBindings.add(String.format("%s:%s", DORIS_QUERY_PORT, DORIS_QUERY_PORT));
Review Comment:
@hailin0 This is reference with doris arch.
When we request fe' node to send data, doris's fe will return a be node for client, then client will redirect to request be again, when client request to be's node will use query port.
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] hk-lrzy commented on a diff in pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
hk-lrzy commented on code in PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#discussion_r969108379
##########
pom.xml:
##########
@@ -198,6 +198,7 @@
<checker.qual.version>3.10.0</checker.qual.version>
<awaitility.version>4.2.0</awaitility.version>
<jwt.version>0.10.7</jwt.version>
+ <httpclient.version>4.5.13</httpclient.version>
Review Comment:
Sure, thanks, i will move this to my connector.
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] ic4y commented on a diff in pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
ic4y commented on code in PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#discussion_r964655219
##########
seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/common/DorisOptions.java:
##########
@@ -0,0 +1,143 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.doris.common;
+
+import org.apache.seatunnel.common.PropertiesUtil;
+
+import org.apache.seatunnel.shade.com.typesafe.config.Config;
+
+import lombok.Builder;
+import lombok.Data;
+
+import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Created 2022/8/01
+ */
Review Comment:
as above
##########
seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/common/HttpClient.java:
##########
@@ -0,0 +1,156 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.doris.common;
+
+import com.google.common.io.Closeables;
+import org.apache.http.HttpEntity;
+import org.apache.http.HttpResponse;
+import org.apache.http.HttpStatus;
+import org.apache.http.client.config.RequestConfig;
+import org.apache.http.client.methods.CloseableHttpResponse;
+import org.apache.http.client.methods.HttpDelete;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.methods.HttpHead;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.client.methods.HttpPut;
+import org.apache.http.client.methods.HttpUriRequest;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.DefaultHttpRequestRetryHandler;
+import org.apache.http.impl.client.DefaultRedirectStrategy;
+import org.apache.http.impl.client.HttpClients;
+import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
+import org.apache.http.util.EntityUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+/**
+ * Created 2022/8/01
+ */
Review Comment:
as above
##########
seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/common/DorisConstants.java:
##########
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.doris.common;
+
+import java.util.Arrays;
+import java.util.List;
+
+/**
+ * Created 2022/8/24
+ */
Review Comment:
Pointless suggest delete it
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] ic4y commented on pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
ic4y commented on PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#issuecomment-1239197100
Need to add spark e2e test
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] TyrantLucifer commented on a diff in pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
TyrantLucifer commented on code in PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#discussion_r966831322
##########
pom.xml:
##########
@@ -198,6 +198,7 @@
<checker.qual.version>3.10.0</checker.qual.version>
<awaitility.version>4.2.0</awaitility.version>
<jwt.version>0.10.7</jwt.version>
+ <httpclient.version>4.5.13</httpclient.version>
Review Comment:
Please not define the dependency that connector needed in main pom, you should define it in your connector own pom.
##########
seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/common/HttpClient.java:
##########
@@ -0,0 +1,153 @@
+/*
Review Comment:
How about use the ability of `http-connector-base` instead of this?
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] TyrantLucifer commented on a diff in pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
TyrantLucifer commented on code in PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#discussion_r970600603
##########
seatunnel-connectors-v2/connector-doris/pom.xml:
##########
@@ -0,0 +1,95 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You under the Apache License, Version 2.0
+ (the "License"); you may not use this file except in compliance with
+ the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <parent>
+ <artifactId>seatunnel-connectors-v2</artifactId>
+ <groupId>org.apache.seatunnel</groupId>
+ <version>${revision}</version>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <artifactId>connector-doris</artifactId>
+ <properties>
+ <httpclient.version>4.5.13</httpclient.version>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>connector-common</artifactId>
+ <version>${revision}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>seatunnel-format-json</artifactId>
+ <version>${revision}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.httpcomponents</groupId>
+ <artifactId>httpclient</artifactId>
+ <version>${httpclient.version}</version>
+ </dependency>
+ </dependencies>
+
+ <build>
Review Comment:
You means that `http-client` conflict with spark's version, right?
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] TyrantLucifer commented on a diff in pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
TyrantLucifer commented on code in PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#discussion_r970592503
##########
seatunnel-e2e/seatunnel-spark-connector-v2-e2e/connector-doris-spark-e2e/src/test/resources/doris/fakesource_to_doris.conf:
##########
@@ -0,0 +1,71 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+######
+###### This config file is a demonstration of streaming processing in seatunnel config
+######
+
+env {
Review Comment:
You can refer to https://github.com/apache/incubator-seatunnel/blob/dev/seatunnel-e2e/seatunnel-spark-connector-v2-e2e/connector-fake-spark-e2e/src/test/resources/fake/fakesource_to_console.conf
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] CalvinKirs commented on pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris.
Posted by GitBox <gi...@apache.org>.
CalvinKirs commented on PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#issuecomment-1233855284
Hi, are there any issues with existing doris images on docker hub?
On 09/1/2022 ***@***.***> wrote：
This situation (because the framework itself is not integrated) requires us to create it manually. FYI : https://www.testcontainers.org/features/creating_images/ On 09/1/2022 @.> wrote： flink Sure, i am try to , but there have no avaliable docker image for doris now. — Reply to this email directly, view it on GitHub, or unsubscribe. You are receiving this because you commented.Message ID: @.>
Thanks, i am already try to build image by myself, and it will push to dockerhub.
—
Reply to this email directly, view it on GitHub, or unsubscribe.
You are receiving this because you commented.Message ID: ***@***.***>
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] hk-lrzy commented on a diff in pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
hk-lrzy commented on code in PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#discussion_r970947243
##########
seatunnel-e2e/seatunnel-spark-connector-v2-e2e/connector-doris-spark-e2e/src/test/resources/doris/fakesource_to_doris.conf:
##########
@@ -0,0 +1,75 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+######
+###### This config file is a demonstration of streaming processing in seatunnel config
+######
+
+env {
+ # You can set flink configuration here
+ job.mode = "BATCH"
+ spark.app.name = "SeaTunnel"
+ spark.executor.instances = 2
+ spark.executor.cores = 1
+ spark.executor.memory = "1g"
+ spark.master = local
+ #execution.checkpoint.interval = 10000
+ #execution.checkpoint.data-uri = "hdfs://localhost:9000/checkpoint"
+}
+
+source {
+ # This is a example source plugin **only for test and demonstrate the feature source plugin**
+ FakeSource {
+ result_table_name = "fake"
+ schema = {
+ fields {
+ user_id = "bigint",
+ city = "string",
+ age = "smallint",
+ sex = "smallint",
+ cost = "bigint",
+ max_dwell_time = "int",
+ min_dwell_time = "int",
+ price = "double"
Review Comment:
And anyother datatype reference with https://github.com/apache/incubator-seatunnel/pull/2586#discussion_r970944700
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] hk-lrzy commented on a diff in pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
hk-lrzy commented on code in PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#discussion_r970824962
##########
seatunnel-e2e/seatunnel-flink-connector-v2-e2e/connector-flink-e2e-base/src/test/resources/log4j.properties:
##########
@@ -0,0 +1,22 @@
+#
Review Comment:
@hailin0 I have difference thought about this one? Why we need copy the duplicate file to those module rather than put in base module?
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] EricJoy2048 commented on pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
EricJoy2048 commented on PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#issuecomment-1300028676
Since there has been no update for a long time, we decided to close this PR. If anyone is willing to complete it, please leave a message and let us know.
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] TaoZex commented on pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
TaoZex commented on PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#issuecomment-1300059001
Please assign to me.
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] hk-lrzy commented on a diff in pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
hk-lrzy commented on code in PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#discussion_r970594665
##########
seatunnel-core/seatunnel-flink-starter/src/main/java/org/apache/seatunnel/core/starter/flink/execution/SinkExecuteProcessor.java:
##########
@@ -75,7 +75,8 @@ public List<DataStream<Row>> execute(List<DataStream<Row>> upstreamDataStreams)
SeaTunnelSink<SeaTunnelRow, Serializable, Serializable, Serializable> seaTunnelSink = plugins.get(i);
DataStream<Row> stream = fromSourceTable(sinkConfig).orElse(input);
seaTunnelSink.setTypeInfo((SeaTunnelRowType) TypeConverterUtils.convert(stream.getType()));
- stream.sinkTo(new FlinkSink<>(seaTunnelSink));
+ stream.sinkTo(new FlinkSink<>(seaTunnelSink)).name(seaTunnelSink.getPluginName())
Review Comment:
https://github.com/apache/incubator-seatunnel/pull/2728 i have this one to fixed this issue.
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] hk-lrzy commented on a diff in pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
hk-lrzy commented on code in PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#discussion_r970584745
##########
seatunnel-connectors-v2/connector-doris/pom.xml:
##########
@@ -0,0 +1,95 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You under the Apache License, Version 2.0
+ (the "License"); you may not use this file except in compliance with
+ the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <parent>
+ <artifactId>seatunnel-connectors-v2</artifactId>
+ <groupId>org.apache.seatunnel</groupId>
+ <version>${revision}</version>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <artifactId>connector-doris</artifactId>
+ <properties>
+ <httpclient.version>4.5.13</httpclient.version>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>connector-common</artifactId>
+ <version>${revision}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>seatunnel-format-json</artifactId>
+ <version>${revision}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.httpcomponents</groupId>
+ <artifactId>httpclient</artifactId>
+ <version>${httpclient.version}</version>
+ </dependency>
+ </dependencies>
+
+ <build>
Review Comment:
Becasue it conflict with spark. you can run the spark e2e testing with the shaded plugin. it will have exception.
And i see the dependency in the spark's lib, the version is `4.5.2`, so we need use shaded to avoid it.
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] hk-lrzy commented on pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris.
Posted by GitBox <gi...@apache.org>.
hk-lrzy commented on PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#issuecomment-1236557148
@CalvinKirs @hailin0 @EricJoy2048 @jbonofre i already added e2e test for doris, plz take a look again
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] hk-lrzy commented on a diff in pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
hk-lrzy commented on code in PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#discussion_r964980234
##########
seatunnel-examples/seatunnel-flink-connector-v2-example/pom.xml:
##########
@@ -32,6 +32,16 @@
<flink.scope>compile</flink.scope>
</properties>
+ <dependencyManagement>
+ <dependencies>
+ <dependency>
+ <groupId>commons-io</groupId>
+ <artifactId>commons-io</artifactId>
+ <version>2.8.0</version>
+ </dependency>
+ </dependencies>
+ </dependencyManagement>
+
Review Comment:
can‘t revert, because there have conflict by this dependency.
it caused by databus import low version `commons-io`
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] hk-lrzy commented on a diff in pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
hk-lrzy commented on code in PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#discussion_r965004841
##########
seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/common/HttpClient.java:
##########
@@ -0,0 +1,153 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.doris.common;
+
+import com.google.common.io.Closeables;
+import org.apache.http.HttpEntity;
+import org.apache.http.HttpResponse;
+import org.apache.http.HttpStatus;
+import org.apache.http.client.config.RequestConfig;
+import org.apache.http.client.methods.CloseableHttpResponse;
+import org.apache.http.client.methods.HttpDelete;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.methods.HttpHead;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.client.methods.HttpPut;
+import org.apache.http.client.methods.HttpUriRequest;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.DefaultHttpRequestRetryHandler;
+import org.apache.http.impl.client.DefaultRedirectStrategy;
+import org.apache.http.impl.client.HttpClients;
+import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
+import org.apache.http.util.EntityUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+public class HttpClient {
+ private static final Logger LOG = LoggerFactory.getLogger(HttpClient.class);
+
+ private CloseableHttpClient client;
+ private PoolingHttpClientConnectionManager connectionManager;
+
+ public HttpClient(DorisOptions dorisOptions) {
Review Comment:
does you mean use jdbc to write doris? i think it's not good iead?
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] hk-lrzy commented on pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris.
Posted by GitBox <gi...@apache.org>.
hk-lrzy commented on PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#issuecomment-1233851356
> This situation (because the framework itself is not integrated) requires us to create it manually. FYI : https://www.testcontainers.org/features/creating_images/ On 09/1/2022 ***@***.***> wrote： flink Sure, i am try to , but there have no avaliable docker image for doris now. — Reply to this email directly, view it on GitHub, or unsubscribe. You are receiving this because you commented.Message ID: ***@***.***>
Thanks, i am already try to build image by myself, and it will push to dockerhub.
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] hk-lrzy commented on pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris.
Posted by GitBox <gi...@apache.org>.
hk-lrzy commented on PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#issuecomment-1234217403
> Thanks, i am already try to build image by myself, and it will push to dockerhub.
> Hi, are there any issues with existing doris images on docker hub? On 09/1/2022 ***@***.***> wrote：
it's just use for build-env, you may see the image's tag like [build-env-for-1.1.0](https://hub.docker.com/layers/apache/doris/build-env-for-1.1.0/images/sha256-174d1f3106c200625155416d16d1db02939ce3f976808b995e7c381d939209ae?context=explore)
it is not for the running pod.
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] hk-lrzy commented on pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris.
Posted by GitBox <gi...@apache.org>.
hk-lrzy commented on PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#issuecomment-1235188419
>
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] hk-lrzy commented on a diff in pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
hk-lrzy commented on code in PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#discussion_r964979005
##########
seatunnel-examples/seatunnel-flink-connector-v2-example/pom.xml:
##########
@@ -76,6 +86,11 @@
<artifactId>connector-dingtalk</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>connector-datahub</artifactId>
+ <version>${project.version}</version>
+ </dependency>
Review Comment:
sure
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] hailin0 commented on a diff in pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
hailin0 commented on code in PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#discussion_r964906630
##########
seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/common/HttpClient.java:
##########
@@ -0,0 +1,153 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.doris.common;
+
+import com.google.common.io.Closeables;
+import org.apache.http.HttpEntity;
+import org.apache.http.HttpResponse;
+import org.apache.http.HttpStatus;
+import org.apache.http.client.config.RequestConfig;
+import org.apache.http.client.methods.CloseableHttpResponse;
+import org.apache.http.client.methods.HttpDelete;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.client.methods.HttpHead;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.client.methods.HttpPut;
+import org.apache.http.client.methods.HttpUriRequest;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.DefaultHttpRequestRetryHandler;
+import org.apache.http.impl.client.DefaultRedirectStrategy;
+import org.apache.http.impl.client.HttpClients;
+import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
+import org.apache.http.util.EntityUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.IOException;
+
+public class HttpClient {
+ private static final Logger LOG = LoggerFactory.getLogger(HttpClient.class);
+
+ private CloseableHttpClient client;
+ private PoolingHttpClientConnectionManager connectionManager;
+
+ public HttpClient(DorisOptions dorisOptions) {
Review Comment:
@hk-lrzy
reference
https://github.com/apache/incubator-seatunnel/pull/2536#discussion_r956830723
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] EricJoy2048 commented on pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
EricJoy2048 commented on PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#issuecomment-1277057834
@hk-lrzy Please fix the CI error.
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] hk-lrzy commented on a diff in pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
hk-lrzy commented on code in PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#discussion_r969111261
##########
seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/common/HttpClient.java:
##########
@@ -0,0 +1,153 @@
+/*
Review Comment:
I perfer we have a new module to maintain those clients, and connectors should dependency those client if necessary. how about this one? @TyrantLucifer
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] hk-lrzy commented on a diff in pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
hk-lrzy commented on code in PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#discussion_r970584745
##########
seatunnel-connectors-v2/connector-doris/pom.xml:
##########
@@ -0,0 +1,95 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You under the Apache License, Version 2.0
+ (the "License"); you may not use this file except in compliance with
+ the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <parent>
+ <artifactId>seatunnel-connectors-v2</artifactId>
+ <groupId>org.apache.seatunnel</groupId>
+ <version>${revision}</version>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <artifactId>connector-doris</artifactId>
+ <properties>
+ <httpclient.version>4.5.13</httpclient.version>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>connector-common</artifactId>
+ <version>${revision}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>seatunnel-format-json</artifactId>
+ <version>${revision}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.httpcomponents</groupId>
+ <artifactId>httpclient</artifactId>
+ <version>${httpclient.version}</version>
+ </dependency>
+ </dependencies>
+
+ <build>
Review Comment:
Becasue it conflict with spark. you can run the spark e2e testing without the shaded plugin, it will have exception.
And i see the dependency in the spark's lib, the version is `4.5.2`, so we need use shaded to avoid it.
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] TyrantLucifer commented on a diff in pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
TyrantLucifer commented on code in PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#discussion_r970604183
##########
seatunnel-e2e/seatunnel-spark-connector-v2-e2e/connector-doris-spark-e2e/src/test/resources/doris/fakesource_to_doris.conf:
##########
@@ -0,0 +1,71 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+######
+###### This config file is a demonstration of streaming processing in seatunnel config
+######
+
+env {
Review Comment:
We should make sure that the configuration file is consistent with the module. In spark module it should not appear flink's parameters.
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] hk-lrzy commented on a diff in pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
hk-lrzy commented on code in PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#discussion_r970611081
##########
seatunnel-connectors-v2/connector-doris/pom.xml:
##########
@@ -0,0 +1,95 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You under the Apache License, Version 2.0
+ (the "License"); you may not use this file except in compliance with
+ the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <parent>
+ <artifactId>seatunnel-connectors-v2</artifactId>
+ <groupId>org.apache.seatunnel</groupId>
+ <version>${revision}</version>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <artifactId>connector-doris</artifactId>
+ <properties>
+ <httpclient.version>4.5.13</httpclient.version>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>connector-common</artifactId>
+ <version>${revision}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>seatunnel-format-json</artifactId>
+ <version>${revision}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.httpcomponents</groupId>
+ <artifactId>httpclient</artifactId>
+ <version>${httpclient.version}</version>
+ </dependency>
+ </dependencies>
+
+ <build>
Review Comment:
> You means that http-client conflict with spark's version, right?
right. and i put the screenshot to show the version difference.
##########
seatunnel-connectors-v2/connector-doris/pom.xml:
##########
@@ -0,0 +1,95 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You under the Apache License, Version 2.0
+ (the "License"); you may not use this file except in compliance with
+ the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <parent>
+ <artifactId>seatunnel-connectors-v2</artifactId>
+ <groupId>org.apache.seatunnel</groupId>
+ <version>${revision}</version>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <artifactId>connector-doris</artifactId>
+ <properties>
+ <httpclient.version>4.5.13</httpclient.version>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>connector-common</artifactId>
+ <version>${revision}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>seatunnel-format-json</artifactId>
+ <version>${revision}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.httpcomponents</groupId>
+ <artifactId>httpclient</artifactId>
+ <version>${httpclient.version}</version>
+ </dependency>
+ </dependencies>
+
+ <build>
Review Comment:
> You means that http-client conflict with spark's version, right?
right. and i put the screenshot to show the version difference.
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] hk-lrzy commented on pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
hk-lrzy commented on PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#issuecomment-1246684405
@TyrantLucifer @EricJoy2048
Plz take a look again, if there have no any more questions, plz approve the pr, thanks a lot.
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] TyrantLucifer commented on a diff in pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
TyrantLucifer commented on code in PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#discussion_r970767330
##########
seatunnel-e2e/seatunnel-spark-connector-v2-e2e/connector-doris-spark-e2e/src/test/resources/doris/fakesource_to_doris.conf:
##########
@@ -0,0 +1,75 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+######
+###### This config file is a demonstration of streaming processing in seatunnel config
+######
+
+env {
+ # You can set flink configuration here
Review Comment:
Spark not flink
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] hk-lrzy commented on a diff in pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
hk-lrzy commented on code in PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#discussion_r970820845
##########
seatunnel-e2e/seatunnel-flink-connector-v2-e2e/connector-doris-flink-e2e/src/test/java/org/apache/seatunnel/e2e/flink/v2/doris/FakeSourceToDorisIT.java:
##########
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.e2e.flink.v2.doris;
+
+import org.apache.seatunnel.e2e.flink.FlinkContainer;
+
+import com.google.common.collect.Lists;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.Container;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.containers.wait.strategy.HostPortWaitStrategy;
+import org.testcontainers.lifecycle.Startables;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
+
+public class FakeSourceToDorisIT extends FlinkContainer {
+ private static final Logger LOG = LoggerFactory.getLogger(FakeSourceToDorisIT.class);
+
+ private static final String DORIS_DRIVER = "com.mysql.cj.jdbc.Driver";
+ private static final String DORIS_CONNECTION_URL = "jdbc:mysql://localhost:9030?rewriteBatchedStatements=true";
+ private static final String DORIS_PASSWD = "";
+ private static final String DORIS_USERNAME = "root";
+
+ private static final String DORIS_DATABASE = "test";
+ private static final String DORIS_TABLE = "seatunnel";
+ private static final String DORIS_DATABASE_DDL = "CREATE DATABASE IF NOT EXISTS `" + DORIS_DATABASE + "`";
+ private static final String DORIS_USE_DATABASE = "USE `" + DORIS_DATABASE + "`";
+ private static final String DORIS_TABLE_DDL = "CREATE TABLE IF NOT EXISTS `" + DORIS_DATABASE + "`.`" + DORIS_TABLE + "` ( " +
+ " `user_id` LARGEINT NOT NULL COMMENT 'id'," +
+ " `date` DATE NOT NULL COMMENT 'date'," +
+ " `city` VARCHAR(20) COMMENT 'city'," +
+ " `age` SMALLINT COMMENT 'age'," +
+ " `sex` TINYINT COMMENT 'sec'," +
+ " `last_visit_date` DATETIME REPLACE DEFAULT '1970-01-01 00:00:00' ," +
+ " `cost` BIGINT SUM DEFAULT '0' ," +
+ " `max_dwell_time` INT MAX DEFAULT '0' ," +
+ " `min_dwell_time` INT MIN DEFAULT '99999'" +
+ ") AGGREGATE KEY(`user_id`, `date`, `city`, `age`, `sex`) DISTRIBUTED BY HASH(`user_id`) BUCKETS 1 PROPERTIES (" +
+ " 'replication_allocation' = 'tag.location.default: 1'" +
+ ");";
+
+ private static final String DORIS_TRUNCATE_TABLE = "TRUNCATE TABLE `" + DORIS_TABLE + "`";
+ private static final String DORIS_SELECT_TABLE = "SELECT COUNT(*) FROM `" + DORIS_TABLE + "`";
+
+ //thanks zhaomin1432 provided the doris images.
+ private static final String DORIS_IMAGE_NAME = "zhaomin1423/doris:1.0.0-b2";
Review Comment:
Doris doesn't have avaliable docker image, previos comment already talk about this one.
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] hk-lrzy commented on a diff in pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
hk-lrzy commented on code in PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#discussion_r970824962
##########
seatunnel-e2e/seatunnel-flink-connector-v2-e2e/connector-flink-e2e-base/src/test/resources/log4j.properties:
##########
@@ -0,0 +1,22 @@
+#
Review Comment:
@hailin0 I have difference thought about this one? Why we need duplicate file to those moudle rather than put in base module?
##########
seatunnel-e2e/seatunnel-flink-connector-v2-e2e/connector-flink-e2e-base/src/test/resources/log4j.properties:
##########
@@ -0,0 +1,22 @@
+#
Review Comment:
@hailin0 I have difference thought about this one? Why we need copy the duplicate file to those moudle rather than put in base module?
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] hk-lrzy commented on a diff in pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
hk-lrzy commented on code in PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#discussion_r970581581
##########
seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/sink/DorisWriter.java:
##########
@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.connectors.doris.sink;
+
+import org.apache.seatunnel.api.sink.SinkWriter;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import org.apache.seatunnel.common.utils.JsonUtils;
+import org.apache.seatunnel.connectors.doris.common.DorisConstants;
+import org.apache.seatunnel.connectors.doris.common.DorisOptions;
+import org.apache.seatunnel.connectors.doris.sink.loader.DorisStreamLoader;
+import org.apache.seatunnel.connectors.seatunnel.common.sink.AbstractSinkWriter;
+import org.apache.seatunnel.format.json.JsonSerializationSchema;
+
+import org.apache.seatunnel.shade.com.typesafe.config.Config;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import org.apache.commons.collections4.CollectionUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.time.Instant;
+import java.time.ZoneId;
+import java.time.format.DateTimeFormatter;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+
+public class DorisWriter extends AbstractSinkWriter<SeaTunnelRow, Void> {
+ private static final Logger LOG = LoggerFactory.getLogger(DorisWriter.class);
+
+ private static final String LABEL_TEMPLATE = "setunnel_sink_subtask_%s_%s_%s";
+
+ private final SeaTunnelRowType seaTunnelRowType;
+ private final DorisOptions options;
+ private final JsonSerializationSchema serializationSchema;
+ private final DorisLoader<String> loader;
+
+ private final List<JsonNode> batch;
+ private final DateTimeFormatter formatter;
+ private final SinkWriter.Context context;
+
+ public DorisWriter(Config dorisSinkConf,
+ SeaTunnelRowType seaTunnelRowType,
+ SinkWriter.Context context) {
+ this.seaTunnelRowType = seaTunnelRowType;
+ this.options = DorisOptions.fromPluginConfig(dorisSinkConf);
+ this.serializationSchema = new JsonSerializationSchema(seaTunnelRowType);
+ //now we only support stream load, maybe future broker load will implement in seatunnel.
+ this.loader = new DorisStreamLoader(options);
+ this.batch = new ArrayList<>();
+ this.formatter = DateTimeFormatter.ofPattern(DorisConstants.DORIS_LABEL_PATTERN_VALUE)
+ .withZone(ZoneId.systemDefault());
Review Comment:
This is not for the date type. This is for the label, so i think it not nessenary to have a new configuration.
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] hk-lrzy commented on a diff in pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
hk-lrzy commented on code in PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#discussion_r970944700
##########
seatunnel-e2e/seatunnel-flink-connector-v2-e2e/connector-doris-flink-e2e/src/test/resources/doris/fakesource_to_doris.conf:
##########
@@ -0,0 +1,73 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+######
+###### This config file is a demonstration of streaming processing in seatunnel config
+######
+
+env {
+ # You can set flink configuration here
+ execution.parallelism = 1
+ job.mode = "BATCH"
+ #execution.checkpoint.interval = 10000
+ #execution.checkpoint.data-uri = "hdfs://localhost:9000/checkpoint"
+}
+
+source {
+ # This is a example source plugin **only for test and demonstrate the feature source plugin**
+ FakeSource {
+ result_table_name = "fake"
+ schema = {
+ fields {
+ user_id = "bigint",
+ date = "date",
+ city = "string",
+ age = "smallint",
+ sex = "smallint",
+ last_visit_date = "timestamp",
+ cost = "bigint",
+ max_dwell_time = "int",
+ min_dwell_time = "int",
+ price = "double"
Review Comment:
@hailin0
https://doris.apache.org/docs/dev/sql-manual/sql-reference/Data-Definition-Statements/Create/CREATE-TABLE/
```
FLOAT (4 bytes)
Support scientific notation
DOUBLE (12 bytes)
Support scientific notation
```
Now `FakeSource`'s range is bigger than doris definetion, so i can't test more data type.
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-seatunnel] hk-lrzy commented on pull request #2586: [SeaTunnel][Connector-v2][Doris]Support Doris Sink.
Posted by GitBox <gi...@apache.org>.
hk-lrzy commented on PR #2586:
URL: https://github.com/apache/incubator-seatunnel/pull/2586#issuecomment-1240674050
@hailin0 I push a commit and resolve most of comments. can you take a look again, thanks a lot.
--
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: commits-unsubscribe@seatunnel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org