This is an automated email from the ASF dual-hosted git repository.
gaoyingju pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/incubator-seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 13028f4 [SeaTunnel #1037] Add InfluxDB Flink Sink (#1038)
13028f4 is described below
commit 13028f4b1d4d035594b77d6d9aa742b4a911cd0d
Author: Benedict Jin <[email protected]>
AuthorDate: Fri Jan 14 12:46:23 2022 +0800
[SeaTunnel #1037] Add InfluxDB Flink Sink (#1038)
---
.../flink/configuration/sink-plugins/InfluxDb.md | 47 +++++++++++
pom.xml | 7 ++
seatunnel-connectors/pom.xml | 2 +-
.../seatunnel-connector-flink-influxdb/pom.xml | 48 +++++++++++
.../seatunnel/flink/sink/InfluxDbOutputFormat.java | 98 ++++++++++++++++++++++
.../apache/seatunnel/flink/sink/InfluxDbSink.java | 74 ++++++++++++++++
.../org.apache.seatunnel.flink.BaseFlinkSink | 1 +
7 files changed, 276 insertions(+), 1 deletion(-)
diff --git a/docs/en/flink/configuration/sink-plugins/InfluxDb.md
b/docs/en/flink/configuration/sink-plugins/InfluxDb.md
new file mode 100644
index 0000000..5523343
--- /dev/null
+++ b/docs/en/flink/configuration/sink-plugins/InfluxDb.md
@@ -0,0 +1,47 @@
+# Sink plugin: InfluxDB
+
+## Description
+
+Write data to InfluxDB.
+
+## Options
+
+| name | type | required | default value |
+| ----------- | -------------- | -------- | ------------- |
+| server_url | `String` | yes | - |
+| database | `String` | yes | - |
+| measurement | `String` | yes | - |
+| tags | `List<String>` | yes | - |
+| fields | `List<String>` | yes | - |
+
+### server_url [`String`]
+
+The URL of InfluxDB Server.
+
+### datasource [`String`]
+
+The DataSource name in InfluxDB.
+
+### measurement [`String`]
+
+The Measurement name in InfluxDB.
+
+### tags [`List<String>`]
+
+The list of Tag in InfluxDB.
+
+### fields [`List<String>`]
+
+The list of Field in InfluxDB.
+
+## Example
+
+```hocon
+InfluxDbSink {
+ server_url = "http://127.0.0.1:8086/";
+ database = "influxdb"
+ measurement = "m"
+ tags = ["country", "city"]
+ fields = ["count"]
+}
+```
diff --git a/pom.xml b/pom.xml
index 7fcaa3b..fe90002 100644
--- a/pom.xml
+++ b/pom.xml
@@ -120,6 +120,7 @@
<config.version>1.3.3</config.version>
<maven-shade-plugin.version>3.2.4</maven-shade-plugin.version>
<maven-helper-plugin.version>3.2.0</maven-helper-plugin.version>
+ <influxdb-java.version>2.22</influxdb-java.version>
</properties>
<dependencyManagement>
@@ -427,6 +428,12 @@
<artifactId>config</artifactId>
<version>${config.version}</version>
</dependency>
+
+ <dependency>
+ <groupId>org.influxdb</groupId>
+ <artifactId>influxdb-java</artifactId>
+ <version>${influxdb-java.version}</version>
+ </dependency>
</dependencies>
</dependencyManagement>
diff --git a/seatunnel-connectors/pom.xml b/seatunnel-connectors/pom.xml
index 619f615..f535191 100644
--- a/seatunnel-connectors/pom.xml
+++ b/seatunnel-connectors/pom.xml
@@ -57,7 +57,7 @@
<module>seatunnel-connector-flink-fake</module>
<module>seatunnel-connector-flink-socket</module>
<module>seatunnel-connector-flink-doris</module>
-
+ <module>seatunnel-connector-flink-influxdb</module>
</modules>
</project>
diff --git a/seatunnel-connectors/seatunnel-connector-flink-influxdb/pom.xml
b/seatunnel-connectors/seatunnel-connector-flink-influxdb/pom.xml
new file mode 100644
index 0000000..92a46aa
--- /dev/null
+++ b/seatunnel-connectors/seatunnel-connector-flink-influxdb/pom.xml
@@ -0,0 +1,48 @@
+<?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>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>seatunnel-connectors</artifactId>
+ <version>2.0.5-SNAPSHOT</version>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <artifactId>seatunnel-connector-flink-influxdb</artifactId>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>seatunnel-api-flink</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-java</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.influxdb</groupId>
+ <artifactId>influxdb-java</artifactId>
+ </dependency>
+ </dependencies>
+</project>
diff --git
a/seatunnel-connectors/seatunnel-connector-flink-influxdb/src/main/java/org/apache/seatunnel/flink/sink/InfluxDbOutputFormat.java
b/seatunnel-connectors/seatunnel-connector-flink-influxdb/src/main/java/org/apache/seatunnel/flink/sink/InfluxDbOutputFormat.java
new file mode 100644
index 0000000..b8ec229
--- /dev/null
+++
b/seatunnel-connectors/seatunnel-connector-flink-influxdb/src/main/java/org/apache/seatunnel/flink/sink/InfluxDbOutputFormat.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.flink.sink;
+
+import org.apache.flink.api.common.io.RichOutputFormat;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.types.Row;
+import org.influxdb.BatchOptions;
+import org.influxdb.InfluxDB;
+import org.influxdb.InfluxDBFactory;
+import org.influxdb.dto.Point;
+import org.influxdb.dto.Query;
+
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+public class InfluxDbOutputFormat extends RichOutputFormat<Row> {
+
+ private final InfluxDB influxDB;
+ private final String measurement;
+ private final List<String> tags;
+ private final List<String> fields;
+
+ public InfluxDbOutputFormat(String serverURL, String database, String
measurement, List<String> tags, List<String> fields) {
+ this.influxDB = InfluxDBFactory.connect(serverURL);
+ this.influxDB.query(new Query("CREATE DATABASE " + database));
+ this.influxDB.setDatabase(database);
+ this.influxDB.enableBatch(
+ BatchOptions.DEFAULTS
+ .threadFactory(runnable -> {
+ Thread thread = new Thread(runnable);
+ thread.setDaemon(true);
+ return thread;
+ })
+ );
+ this.measurement = measurement;
+ this.tags = tags;
+ this.fields = fields;
+ }
+
+ @Override
+ public void open(int taskNumber, int numTasks) {
+ }
+
+ @Override
+ public void configure(Configuration parameters) {
+ }
+
+ @Override
+ public void writeRecord(Row element) {
+ Point.Builder builder = Point.measurement(this.measurement);
+ builder.time(Long.valueOf(element.getField(0).toString()),
TimeUnit.MILLISECONDS);
+ for (int i = 1; i < tags.size() + 1; i++) {
+ Object v = element.getField(i);
+ if (v != null) {
+ builder.tag(tags.get(i - 1), String.valueOf(v));
+ }
+ }
+ for (int i = tags.size() + 1; i < element.getArity(); i++) {
+ Object v = element.getField(i);
+ if (v != null) {
+ if (v instanceof Number) {
+ builder.addField(fields.get(i - 1), (Number) v);
+ } else if (v instanceof String) {
+ builder.addField(fields.get(i - 1), (String) v);
+ } else if (v instanceof Boolean) {
+ builder.addField(fields.get(i - 1), (Boolean) v);
+ } else {
+ throw new RuntimeException("Not support type of field: " +
v);
+ }
+ }
+ }
+ Point point = builder.build();
+ influxDB.write(point);
+ }
+
+ @Override
+ public void close() {
+ if (this.influxDB != null) {
+ this.influxDB.close();
+ }
+ }
+}
diff --git
a/seatunnel-connectors/seatunnel-connector-flink-influxdb/src/main/java/org/apache/seatunnel/flink/sink/InfluxDbSink.java
b/seatunnel-connectors/seatunnel-connector-flink-influxdb/src/main/java/org/apache/seatunnel/flink/sink/InfluxDbSink.java
new file mode 100644
index 0000000..c447ab9
--- /dev/null
+++
b/seatunnel-connectors/seatunnel-connector-flink-influxdb/src/main/java/org/apache/seatunnel/flink/sink/InfluxDbSink.java
@@ -0,0 +1,74 @@
+/*
+ * 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.flink.sink;
+
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.operators.DataSink;
+import org.apache.flink.types.Row;
+import org.apache.seatunnel.common.config.CheckConfigUtil;
+import org.apache.seatunnel.common.config.CheckResult;
+import org.apache.seatunnel.shade.com.typesafe.config.Config;
+import org.apache.seatunnel.flink.FlinkEnvironment;
+import org.apache.seatunnel.flink.batch.FlinkBatchSink;
+
+import java.util.List;
+
+public class InfluxDbSink implements FlinkBatchSink<Row, Row> {
+
+ private static final String SERVER_URL = "server_url";
+ private static final String DATABASE = "database";
+ private static final String MEASUREMENT = "measurement";
+ private static final String TAGS = "tags";
+ private static final String FIELDS = "fields";
+
+ private Config config;
+ private String serverURL;
+ private String database;
+ private String measurement;
+ private List<String> tags;
+ private List<String> fields;
+
+ @Override
+ public DataSink<Row> outputBatch(FlinkEnvironment env, DataSet<Row>
dataSet) {
+ return dataSet.output(new InfluxDbOutputFormat(serverURL, database,
measurement, tags, fields));
+ }
+
+ @Override
+ public void setConfig(Config config) {
+ this.config = config;
+ }
+
+ @Override
+ public Config getConfig() {
+ return config;
+ }
+
+ @Override
+ public CheckResult checkConfig() {
+ return CheckConfigUtil.check(config, SERVER_URL, DATABASE,
MEASUREMENT, TAGS, FIELDS);
+ }
+
+ @Override
+ public void prepare(FlinkEnvironment env) {
+ this.serverURL = config.getString(SERVER_URL);
+ this.database = config.getString(DATABASE);
+ this.measurement = config.getString(MEASUREMENT);
+ this.tags = config.getStringList(TAGS);
+ this.fields = config.getStringList(FIELDS);
+ }
+}
diff --git
a/seatunnel-connectors/seatunnel-connector-flink-influxdb/src/main/resources/META-INF/services/org.apache.seatunnel.flink.BaseFlinkSink
b/seatunnel-connectors/seatunnel-connector-flink-influxdb/src/main/resources/META-INF/services/org.apache.seatunnel.flink.BaseFlinkSink
new file mode 100644
index 0000000..253c633
--- /dev/null
+++
b/seatunnel-connectors/seatunnel-connector-flink-influxdb/src/main/resources/META-INF/services/org.apache.seatunnel.flink.BaseFlinkSink
@@ -0,0 +1 @@
+org.apache.seatunnel.flink.sink.InfluxDbSink
