This is an automated email from the ASF dual-hosted git repository.
dockerzhang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-inlong.git
The following commit(s) were added to refs/heads/master by this push:
new bfc3fd24a [INLONG-4383][Sort] Add MySQL data load support (#4393)
bfc3fd24a is described below
commit bfc3fd24a745c9491f635363343ccb2b303a4b17
Author: yunqingmoswu <[email protected]>
AuthorDate: Thu May 26 20:44:49 2022 +0800
[INLONG-4383][Sort] Add MySQL data load support (#4393)
---
.../apache/inlong/sort/protocol/node/LoadNode.java | 2 +
.../org/apache/inlong/sort/protocol/node/Node.java | 2 +
.../sort/protocol/node/load/MySqlLoadNode.java | 102 +++++++++++++++++
.../sort/protocol/node/load/MySqlLoadNodeTest.java | 46 ++++++++
.../inlong/sort/parser/MySqlLoadSqlParseTest.java | 124 +++++++++++++++++++++
5 files changed, 276 insertions(+)
diff --git
a/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/LoadNode.java
b/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/LoadNode.java
index eb9804664..98d3324b8 100644
---
a/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/LoadNode.java
+++
b/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/LoadNode.java
@@ -33,6 +33,7 @@ import
org.apache.inlong.sort.protocol.node.load.FileSystemLoadNode;
import org.apache.inlong.sort.protocol.node.load.HbaseLoadNode;
import org.apache.inlong.sort.protocol.node.load.HiveLoadNode;
import org.apache.inlong.sort.protocol.node.load.KafkaLoadNode;
+import org.apache.inlong.sort.protocol.node.load.MySqlLoadNode;
import org.apache.inlong.sort.protocol.node.load.PostgresLoadNode;
import org.apache.inlong.sort.protocol.node.load.SqlServerLoadNode;
import org.apache.inlong.sort.protocol.node.load.TDSQLPostgresLoadNode;
@@ -59,6 +60,7 @@ import java.util.Map;
@JsonSubTypes.Type(value = ClickHouseLoadNode.class, name =
"clickHouseLoad"),
@JsonSubTypes.Type(value = SqlServerLoadNode.class, name =
"sqlserverLoad"),
@JsonSubTypes.Type(value = TDSQLPostgresLoadNode.class, name =
"tdsqlPostgresLoad"),
+ @JsonSubTypes.Type(value = MySqlLoadNode.class, name = "mysqlLoad")
})
@NoArgsConstructor
@Data
diff --git
a/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/Node.java
b/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/Node.java
index 44cc50098..470901782 100644
---
a/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/Node.java
+++
b/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/Node.java
@@ -35,6 +35,7 @@ import
org.apache.inlong.sort.protocol.node.load.FileSystemLoadNode;
import org.apache.inlong.sort.protocol.node.load.HbaseLoadNode;
import org.apache.inlong.sort.protocol.node.load.HiveLoadNode;
import org.apache.inlong.sort.protocol.node.load.KafkaLoadNode;
+import org.apache.inlong.sort.protocol.node.load.MySqlLoadNode;
import org.apache.inlong.sort.protocol.node.load.PostgresLoadNode;
import org.apache.inlong.sort.protocol.node.load.SqlServerLoadNode;
import org.apache.inlong.sort.protocol.node.load.TDSQLPostgresLoadNode;
@@ -72,6 +73,7 @@ import java.util.TreeMap;
@JsonSubTypes.Type(value = ClickHouseLoadNode.class, name =
"clickHouseLoad"),
@JsonSubTypes.Type(value = SqlServerLoadNode.class, name =
"sqlserverLoad"),
@JsonSubTypes.Type(value = TDSQLPostgresLoadNode.class, name =
"tdsqlPostgresLoad"),
+ @JsonSubTypes.Type(value = MySqlLoadNode.class, name = "mysqlLoad")
})
public interface Node {
diff --git
a/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/load/MySqlLoadNode.java
b/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/load/MySqlLoadNode.java
new file mode 100644
index 000000000..9ed44d7dc
--- /dev/null
+++
b/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/load/MySqlLoadNode.java
@@ -0,0 +1,102 @@
+/*
+ * 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.inlong.sort.protocol.node.load;
+
+import com.google.common.base.Preconditions;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.NoArgsConstructor;
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonCreator;
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+import
org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonTypeName;
+import org.apache.inlong.sort.protocol.FieldInfo;
+import org.apache.inlong.sort.protocol.enums.FilterStrategy;
+import org.apache.inlong.sort.protocol.node.LoadNode;
+import org.apache.inlong.sort.protocol.transformation.FieldRelationShip;
+import org.apache.inlong.sort.protocol.transformation.FilterFunction;
+
+import java.io.Serializable;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * MySql load node can load data into MySql
+ */
+@EqualsAndHashCode(callSuper = true)
+@JsonTypeName("mysqlLoad")
+@Data
+@NoArgsConstructor
+public class MySqlLoadNode extends LoadNode implements Serializable {
+ /**
+ * jdbc:mysql://host:port/database
+ */
+ @JsonProperty("url")
+ private String url;
+ @JsonProperty("username")
+ private String username;
+ @JsonProperty("password")
+ private String password;
+ @JsonProperty("tableName")
+ private String tableName;
+ @JsonProperty("primaryKey")
+ private String primaryKey;
+
+ @JsonCreator
+ public MySqlLoadNode(
+ @JsonProperty("id") String id,
+ @JsonProperty("name") String name,
+ @JsonProperty("fields") List<FieldInfo> fields,
+ @JsonProperty("fieldRelationShips") List<FieldRelationShip>
fieldRelationShips,
+ @JsonProperty("filters") List<FilterFunction> filters,
+ @JsonProperty("filterStrategy") FilterStrategy filterStrategy,
+ @JsonProperty("sinkParallelism") Integer sinkParallelism,
+ @JsonProperty("properties") Map<String, String> properties,
+ @JsonProperty("url") String url,
+ @JsonProperty("username") String username,
+ @JsonProperty("password") String password,
+ @JsonProperty("tableName") String tableName,
+ @JsonProperty("primaryKey") String primaryKey) {
+ super(id, name, fields, fieldRelationShips, filters, filterStrategy,
sinkParallelism, properties);
+ this.url = Preconditions.checkNotNull(url, "url is null");
+ this.username = Preconditions.checkNotNull(username, "username is
null");
+ this.password = Preconditions.checkNotNull(password, "password is
null");
+ this.tableName = Preconditions.checkNotNull(tableName, "tableName is
null");
+ this.primaryKey = primaryKey;
+ }
+
+ @Override
+ public Map<String, String> tableOptions() {
+ Map<String, String> options = super.tableOptions();
+ options.put("connector", "jdbc-inlong");
+ options.put("url", url);
+ options.put("username", username);
+ options.put("password", password);
+ options.put("table-name", tableName);
+ return options;
+ }
+
+ @Override
+ public String genTableName() {
+ return String.format("table_%s", super.getId());
+ }
+
+ @Override
+ public String getPrimaryKey() {
+ return primaryKey;
+ }
+}
diff --git
a/inlong-sort/sort-common/src/test/java/org/apache/inlong/sort/protocol/node/load/MySqlLoadNodeTest.java
b/inlong-sort/sort-common/src/test/java/org/apache/inlong/sort/protocol/node/load/MySqlLoadNodeTest.java
new file mode 100644
index 000000000..83e47a8f9
--- /dev/null
+++
b/inlong-sort/sort-common/src/test/java/org/apache/inlong/sort/protocol/node/load/MySqlLoadNodeTest.java
@@ -0,0 +1,46 @@
+/*
+ * 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.inlong.sort.protocol.node.load;
+
+import org.apache.inlong.sort.SerializeBaseTest;
+import org.apache.inlong.sort.formats.common.StringFormatInfo;
+import org.apache.inlong.sort.protocol.FieldInfo;
+import org.apache.inlong.sort.protocol.transformation.FieldRelationShip;
+
+import java.util.Collections;
+
+/**
+ * Test for {@link MySqlLoadNode}
+ */
+public class MySqlLoadNodeTest extends SerializeBaseTest<MySqlLoadNode> {
+
+ @Override
+ public MySqlLoadNode getTestObject() {
+ return new MySqlLoadNode("1", "mysql_output",
+ Collections.singletonList(new FieldInfo("name", new
StringFormatInfo())),
+ Collections.singletonList(new FieldRelationShip(new
FieldInfo("name",
+ new StringFormatInfo()), new FieldInfo("name", new
StringFormatInfo()))),
+ null, null, 1, null,
+ "jdbc:mysql://localhost:3306/inlong",
+ "inlong",
+ "inlong",
+ "user",
+ "name");
+ }
+}
diff --git
a/inlong-sort/sort-core/src/test/java/org/apache/inlong/sort/parser/MySqlLoadSqlParseTest.java
b/inlong-sort/sort-core/src/test/java/org/apache/inlong/sort/parser/MySqlLoadSqlParseTest.java
new file mode 100644
index 000000000..20f944cb9
--- /dev/null
+++
b/inlong-sort/sort-core/src/test/java/org/apache/inlong/sort/parser/MySqlLoadSqlParseTest.java
@@ -0,0 +1,124 @@
+/*
+ * 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.inlong.sort.parser;
+
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.table.api.EnvironmentSettings;
+import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
+import org.apache.flink.test.util.AbstractTestBase;
+import org.apache.inlong.sort.formats.common.IntFormatInfo;
+import org.apache.inlong.sort.formats.common.LongFormatInfo;
+import org.apache.inlong.sort.formats.common.StringFormatInfo;
+import org.apache.inlong.sort.parser.impl.FlinkSqlParser;
+import org.apache.inlong.sort.parser.result.ParseResult;
+import org.apache.inlong.sort.protocol.FieldInfo;
+import org.apache.inlong.sort.protocol.GroupInfo;
+import org.apache.inlong.sort.protocol.StreamInfo;
+import org.apache.inlong.sort.protocol.node.Node;
+import org.apache.inlong.sort.protocol.node.extract.MySqlExtractNode;
+import org.apache.inlong.sort.protocol.node.load.MySqlLoadNode;
+import org.apache.inlong.sort.protocol.transformation.FieldRelationShip;
+import
org.apache.inlong.sort.protocol.transformation.relation.NodeRelationShip;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * Test for {@link MySqlLoadNode}
+ */
+public class MySqlLoadSqlParseTest extends AbstractTestBase {
+
+ private MySqlExtractNode buildMySQLExtractNode() {
+ List<FieldInfo> fields = Arrays.asList(new FieldInfo("id", new
LongFormatInfo()),
+ new FieldInfo("name", new StringFormatInfo()),
+ new FieldInfo("age", new IntFormatInfo()));
+ Map<String, String> map = new HashMap<>();
+ return new MySqlExtractNode("1", "mysql_input", fields,
+ null, map, "id",
+ Collections.singletonList("table_input"), "localhost",
"inlong", "inlong",
+ "inlong", null, null,
+ null, null);
+ }
+
+ private Node buildMysqlLoadNode() {
+ List<FieldInfo> fields = Arrays.asList(new FieldInfo("id", new
LongFormatInfo()),
+ new FieldInfo("name", new StringFormatInfo()),
+ new FieldInfo("age", new IntFormatInfo())
+ );
+ List<FieldRelationShip> relations = Arrays
+ .asList(new FieldRelationShip(new FieldInfo("id", new
LongFormatInfo()),
+ new FieldInfo("id", new LongFormatInfo())),
+ new FieldRelationShip(new FieldInfo("name", new
StringFormatInfo()),
+ new FieldInfo("name", new StringFormatInfo())),
+ new FieldRelationShip(new FieldInfo("age", new
IntFormatInfo()),
+ new FieldInfo("age", new IntFormatInfo()))
+ );
+ return new MySqlLoadNode("2", "mysql_output", fields, relations, null,
+ null, null, null, "jdbc:mysql://localhost:3306/inlong",
+ "inlong", "inlong", "table_output", "id");
+ }
+
+ /**
+ * build node relation
+ *
+ * @param inputs extract node
+ * @param outputs load node
+ * @return node relation
+ */
+ private NodeRelationShip buildNodeRelation(List<Node> inputs, List<Node>
outputs) {
+ List<String> inputIds =
inputs.stream().map(Node::getId).collect(Collectors.toList());
+ List<String> outputIds =
outputs.stream().map(Node::getId).collect(Collectors.toList());
+ return new NodeRelationShip(inputIds, outputIds);
+ }
+
+ /**
+ * Test flink sql task for extract is mysql {@link MySqlExtractNode} and
load is mysql {@link MySqlLoadNode}
+ *
+ * @throws Exception The exception may be thrown when executing
+ */
+ @Test
+ public void testMySqlLoadSqlParse() throws Exception {
+ StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setParallelism(1);
+ env.enableCheckpointing(10000);
+ env.disableOperatorChaining();
+ EnvironmentSettings settings = EnvironmentSettings
+ .newInstance()
+ .useBlinkPlanner()
+ .inStreamingMode()
+ .build();
+ StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env,
settings);
+ Node inputNode = buildMySQLExtractNode();
+ Node outputNode = buildMysqlLoadNode();
+ StreamInfo streamInfo = new StreamInfo("1", Arrays.asList(inputNode,
outputNode),
+
Collections.singletonList(buildNodeRelation(Collections.singletonList(inputNode),
+ Collections.singletonList(outputNode))));
+ GroupInfo groupInfo = new GroupInfo("1",
Collections.singletonList(streamInfo));
+ FlinkSqlParser parser = FlinkSqlParser.getInstance(tableEnv,
groupInfo);
+ ParseResult result = parser.parse();
+ Assert.assertTrue(result.tryExecute());
+ }
+
+}
