You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@inlong.apache.org by GitBox <gi...@apache.org> on 2022/09/03 11:10:42 UTC
[GitHub] [inlong] liangyepianzhou opened a new pull request, #5774: [INLONG_5133][Sort] Add Apache InfluxDB Extract and Load Node for Sort
liangyepianzhou opened a new pull request, #5774:
URL: https://github.com/apache/inlong/pull/5774
### Prepare a Pull Request
- Fixes [#XYZ](https://github.com/apache/inlong/issues/5133)
### Motivation
Add Apache InfluxDB Extract and Load Node for Sort
### Modifications
The details can be found at https://github.com/apache/inlong/issues/5133.
### Verifying this change
*(Please pick either of the following options)*
- [ ] This change is a trivial rework/code cleanup without any test coverage.
- [ ] This change is already covered by existing tests, such as:
*(please describe tests)*
- [ ] This change added tests and can be verified as follows:
*(example:)*
- *Added integration tests for end-to-end deployment with large payloads (10MB)*
- *Extended integration test for recovery after broker failure*
### Documentation
- Does this pull request introduces a new feature? (yes / no)
- If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)
--
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@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] yunqingmoswu commented on a diff in pull request #5774: [INLONG-5133][Sort] Support Apache InfluxDB source
Posted by GitBox <gi...@apache.org>.
yunqingmoswu commented on code in PR #5774:
URL: https://github.com/apache/inlong/pull/5774#discussion_r972586581
##########
inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/ExtractNode.java:
##########
@@ -62,7 +63,8 @@
@JsonSubTypes.Type(value = TubeMQExtractNode.class, name = "tubeMQExtract"),
@JsonSubTypes.Type(value = PulsarExtractNode.class, name = "pulsarExtract"),
@JsonSubTypes.Type(value = RedisExtractNode.class, name = "redisExtract"),
- @JsonSubTypes.Type(value = DorisExtractNode.class, name = "dorisExtract")
+ @JsonSubTypes.Type(value = DorisExtractNode.class, name = "dorisExtract"),
+ @JsonSubTypes.Type(value = InfluxDBExtractNode.class, name = "influxDBExtract")
Review Comment:
Please keep the naming consistent of influxExtract between the Node and ExtractNode.
##########
inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/Node.java:
##########
@@ -75,6 +76,7 @@
@JsonSubTypes.Type(value = TubeMQExtractNode.class, name = "tubeMQExtract"),
@JsonSubTypes.Type(value = RedisExtractNode.class, name = "redisExtract"),
@JsonSubTypes.Type(value = DorisExtractNode.class, name = "dorisExtract"),
+ @JsonSubTypes.Type(value = InfluxDBExtractNode.class, name = "influxExtract"),
Review Comment:
Please keep the naming consistent of `influxExtract` between the Node and ExtractNode.
##########
inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/protocol/node/extract/InfluxDBExtractNode.java:
##########
@@ -0,0 +1,115 @@
+/*
+ * 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.extract;
+
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+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.common.enums.MetaField;
+import org.apache.inlong.sort.protocol.FieldInfo;
+import org.apache.inlong.sort.protocol.InlongMetric;
+import org.apache.inlong.sort.protocol.Metadata;
+import org.apache.inlong.sort.protocol.node.ExtractNode;
+import org.apache.inlong.sort.protocol.transformation.WatermarkField;
+
+import javax.annotation.Nullable;
+import java.io.Serializable;
+import java.util.EnumSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * Extract node for influxDB, note that InfluxDB should work in replicaSet mode
+ */
+@EqualsAndHashCode(callSuper = true)
+@JsonTypeName("influxDBExtract")
Review Comment:
Please keep the naming consistent of influxExtract between the Node and ExtractNode.
--
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@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] EMsnap commented on pull request #5774: [INLONG-5133][Sort] Add Apache InfluxDB Extract and Load Node for Sort
Posted by GitBox <gi...@apache.org>.
EMsnap commented on PR #5774:
URL: https://github.com/apache/inlong/pull/5774#issuecomment-1236482706
> > @dockerzhang Please help check this PR when you have time, THX. In addition, I noticed that except for MySQL and JDBC, other CDCs only implement source and not sink. Influx DB also only needs to implement source, right?
>
> I guess this issue means you need to implement both sink and source
also the sink of these connector is implement by jdbc driver, can check the jdbc connector for details
--
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@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] liangyepianzhou commented on pull request #5774: [INLONG-5133][Sort] Add Apache InfluxDB Extract and Load Node for Sort
Posted by GitBox <gi...@apache.org>.
liangyepianzhou commented on PR #5774:
URL: https://github.com/apache/inlong/pull/5774#issuecomment-1236101237
@dockerzhang Please help check this PR when you have time, THX.
In addition, I noticed that except for MySQL and JDBC, other CDCs only implement source and not sink. Influx DB also only needs to implement source, 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@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] EMsnap commented on pull request #5774: [INLONG-5133][Sort] Add Apache InfluxDB Extract and Load Node for Sort
Posted by GitBox <gi...@apache.org>.
EMsnap commented on PR #5774:
URL: https://github.com/apache/inlong/pull/5774#issuecomment-1236480006
> @dockerzhang Please help check this PR when you have time, THX. In addition, I noticed that except for MySQL and JDBC, other CDCs only implement source and not sink. Influx DB also only needs to implement source, right?
I guess this issue means you need to implement both sink and source
--
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@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] github-actions[bot] commented on pull request #5774: [INLONG-5133][Sort] Support InfluxDB extract node
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #5774:
URL: https://github.com/apache/inlong/pull/5774#issuecomment-1496811199
This PR is stale because it has been open for 60 days with no activity.
--
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@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] dockerzhang closed pull request #5774: [INLONG-5133][Sort] Support InfluxDB extract node
Posted by "dockerzhang (via GitHub)" <gi...@apache.org>.
dockerzhang closed pull request #5774: [INLONG-5133][Sort] Support InfluxDB extract node
URL: https://github.com/apache/inlong/pull/5774
--
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@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] liangyepianzhou commented on pull request #5774: [INLONG-5133][Sort] Add Apache InfluxDB Extract and Load Node for Sort
Posted by GitBox <gi...@apache.org>.
liangyepianzhou commented on PR #5774:
URL: https://github.com/apache/inlong/pull/5774#issuecomment-1236100125
@jun0315 I have finished writing the main framework of influx source, and now there are three points that need to be improved. I've listed it below, please take a look at this question when you have time.
1. Parameters
* The parameters written in the building function in the `InfluxDBSource` may not be complete. So if you need to add parameters here, you also need to add them in the `getScanRuntimeProvider` and `InfluxDBTableSourceFactory`.
* Divide required and optional parameters into the methods `requiredOptions` and `optionalOptions`
2. Serialization
* InfluxDB Table Source needs to pass in a `deserializer DebeziumDeserializationSchema<RowData>` deserializer when calling the function; This can get help at [flink-connector-influxdb2](https://github.com/apache/bahir-flink/tree/master/flink-connector-influxdb2)
3. Connecter CanonicalName: There is a configuration item in InfluxDBSource that I don't know how to deal with
` props.setProperty("connector.class", InfluxDbConnector.class.getCanonicalName());`
In order to better understand the workflow of the Flink influx DB source, I wrote a simple process described below:
Workflow of Flink source connecter (simplified version):
`InfluxDBTableSourceFactory` creates an `InfluxDBTableSource`, and the `getScanRuntimeProvider` method of `InfluxDBTableSource` creates a `DebeziumSourceFunction` responsible for fetching captured change data from the database into Flink.
And InfluxDBSource is similar to a creator used to create a DebeziumSourceFunction.
And more details can be found [here](https://nightlies.apache.org/flink/flink-docs-release-1.14/zh/docs/dev/table/sourcessinks/).
--
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@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [inlong] github-actions[bot] commented on pull request #5774: [INLONG-5133][Sort] Support InfluxDB extract node
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on PR #5774:
URL: https://github.com/apache/inlong/pull/5774#issuecomment-1374346941
This PR is stale because it has been open for 60 days with no activity.
--
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@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org