[Feature][Connector-V2][Elasticsearch] Elasticsearch supports spaced write and is relatively real-time data source friendly #4347 #4349

zhilinli123 · 2023-03-14T03:17:30Z

…write and is relatively real-time data source friendly #4347

Purpose of this pull request

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
If necessary, please update the documentation to describe the new feature. https://github.com/apache/incubator-seatunnel/tree/dev/docs
If you are contributing the connector code, please check that the following files are updated:
1. Update change log that in connector document. For more details you can refer to connector-v2
2. Update plugin-mapping.properties and add new connector information in it
3. Update the pom file of seatunnel-dist
Update the release-note.

…write and is relatively real-time data source friendly apache#4347

Hisoka-X · 2023-03-15T02:16:41Z

...va/org/apache/seatunnel/connectors/seatunnel/elasticsearch/sink/ElasticsearchSinkWriter.java

+                        () -> {
+                            synchronized (ElasticsearchSinkWriter.this) {
+                                if (requestEsList.size() > 0 && !isClose) {
+                                    log.info(


Move log into bulkEsWithRetry

Hisoka-X · 2023-03-15T02:16:48Z

...va/org/apache/seatunnel/connectors/seatunnel/elasticsearch/sink/ElasticsearchSinkWriter.java

        if (requestEsList.size() >= maxBatchSize) {
+            log.info("Batch write completion row :" + requestEsList.size());


Move log into bulkEsWithRetry

Hisoka-X · 2023-03-15T02:33:20Z

...va/org/apache/seatunnel/connectors/seatunnel/elasticsearch/sink/ElasticsearchSinkWriter.java

+    private transient ScheduledExecutorService scheduler;
+    private transient ScheduledFuture<?> scheduledFuture;
+    // Whether pre-initialization is required
+    private transient boolean isOpen;


This field is not necessary

Can you point out the number of rows boss?

Hisoka-X · 2023-03-15T02:36:04Z

...va/org/apache/seatunnel/connectors/seatunnel/elasticsearch/sink/ElasticsearchSinkWriter.java

@@ -89,11 +104,50 @@ public void write(SeaTunnelRow element) {

        String indexRequestRow = seaTunnelRowSerializer.serializeRow(element);
        requestEsList.add(indexRequestRow);
+        // Initialize the interval flush
+        tryOpen();


move it into constructer

…write and is relatively real-time data source friendly apache#4347

iture123 · 2023-03-19T12:55:41Z

...va/org/apache/seatunnel/connectors/seatunnel/elasticsearch/sink/ElasticsearchSinkWriter.java

@@ -116,6 +157,9 @@ public synchronized void bulkEsWithRetry(
                                        ElasticsearchConnectorErrorCode.BULK_RESPONSE_ERROR,
                                        "bulk es error: " + bulkResponse.getResponse());
                            }
+                            log.info(


I support not to introduce ScheduledThreadPool,but run Thead.sleep(batchIntervalMs) after bulk request.

I think it's going to be a little bit more write friendly for real-time programs and even if the consumption data is less than the current batch it's going to flush to es
CC @Hisoka-X

Seem like don't a better way to flush data without ScheduledTask. run Thead.sleep(batchIntervalMs) after bulk request will lead data flush more slower, not more faster.

iture123 · 2023-03-19T12:56:49Z

...src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/config/SinkConfig.java

+    public static final Option<Integer> BATCH_INTERVAL_MS =
+            Options.key("batch_interval_ms")
+                    .intType()
+                    .defaultValue(1000)


It is better set defualtValue is 0 ?

It is better set defualtValue is 0 ?
For this time setting, I refer to flink I think it's a little more user friendly

Yes, better to zero, mean close this feature. Only batch_interval_ms bigger than zero, you can create ScheduledTask

You mean that I set batch interval ms to 0 by default? @Hisoka-X

Yes, 0 mean close this feature by default. If user want to open it, just config it.

done

I have resubmitted the default disable policy, which requires the user to manually configure the enable policy
@Hisoka-X

zhilinli123 · 2023-03-22T07:22:45Z

…he#4370

hailin0 · 2023-03-27T08:51:28Z

...src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/config/SinkConfig.java

+    public static final Option<Integer> BATCH_INTERVAL_MS =
+            Options.key("batch_interval_ms")
+                    .intType()
+                    .defaultValue(-1)


Why do you need a default

Why do you need a default
We do not enable it by default until the user manually configures it

hailin0 · 2023-03-27T08:52:10Z

...ain/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/sink/ElasticsearchSink.java

@@ -51,6 +54,8 @@

    private int maxRetryCount = MAX_RETRY_COUNT.defaultValue();

+    private int batchIntervalMs = BATCH_INTERVAL_MS.defaultValue();


Suggested change

private int batchIntervalMs = BATCH_INTERVAL_MS.defaultValue();

private Integer batchIntervalMs = BATCH_INTERVAL_MS.defaultValue();

hailin0 · 2023-03-27T08:54:29Z

...va/org/apache/seatunnel/connectors/seatunnel/elasticsearch/sink/ElasticsearchSinkWriter.java

@@ -94,6 +112,32 @@ public void write(SeaTunnelRow element) {
        }
    }

+    public void open() {


Suggested change

public void open() {

public void startScheduler() {

please examine @hailin0

…write and is relatively real-time data source friendly apache#4347 apache#4349

# Conflicts: # seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/sink/ElasticsearchSink.java # seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/sink/ElasticsearchSinkWriter.java

Hisoka-X

-1.
I don't think it is a good way to add additional threads to write data in the sink. This implementation method makes resources uncontrollable, and also loses the original intention of the committer interface design.

zhilinli123 · 2023-05-05T05:53:33Z

-1. I don't think it is a good way to add additional threads to write data in the sink. This implementation method makes resources uncontrollable, and also loses the original intention of the committer interface design.

boss Do you have any good ideas?

TyrantLucifer · 2023-05-05T08:41:15Z

It's not a good change, for now we do not have the strong need to change it. So I will close it, if you have any questions, please send an e-mail to [email protected] and discuss with community about your solution and your concern.

zhilinli and others added 3 commits March 14, 2023 11:12

[Feature][Connector-V2][Elasticsearch] Elasticsearch supports spaced …

0200ea8

…write and is relatively real-time data source friendly apache#4347

Merge branch 'apache:dev' into feature-flush-es-interval

624d570

[Feature][Connector-V2][Elasticsearch] Elasticsearch supports spaced …

8d3dd6f

…write and is relatively real-time data source friendly apache#4347

Hisoka-X reviewed Mar 15, 2023

View reviewed changes

[Feature][Connector-V2][Elasticsearch] Elasticsearch supports spaced …

aad46d8

…write and is relatively real-time data source friendly apache#4347

iture123 suggested changes Mar 19, 2023

View reviewed changes

[Bug] [connector-v2] [Eleasticsearch] document description error apac…

d3b73d5

…he#4370

zhilinli123 closed this Mar 22, 2023

zhilinli123 reopened this Mar 22, 2023

[Bug] [connector-v2] [Eleasticsearch] document description error apac…

d7461e9

…he#4370

hailin0 reviewed Mar 27, 2023

View reviewed changes

[Feature][Connector-V2][Elasticsearch] Elasticsearch supports spaced …

89636df

…write and is relatively real-time data source friendly apache#4347 apache#4349

zhilinli123 closed this Mar 31, 2023

zhilinli123 reopened this Mar 31, 2023

Hisoka-X reviewed May 5, 2023

View reviewed changes

TyrantLucifer closed this May 5, 2023

TyrantLucifer mentioned this pull request May 5, 2023

[Feature][Connector-V2][Clickhouse] Clickhouse supports spaced write and is relatively real-time data source friendly #4328

Closed

5 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Feature][Connector-V2][Elasticsearch] Elasticsearch supports spaced write and is relatively real-time data source friendly #4347 #4349

[Feature][Connector-V2][Elasticsearch] Elasticsearch supports spaced write and is relatively real-time data source friendly #4347 #4349

zhilinli123 commented Mar 14, 2023

Hisoka-X Mar 15, 2023

Hisoka-X Mar 15, 2023

Hisoka-X Mar 15, 2023

zhilinli123 Mar 15, 2023

Hisoka-X Mar 15, 2023

zhilinli123 Mar 15, 2023

iture123 Mar 19, 2023

zhilinli123 Mar 20, 2023

Hisoka-X Mar 22, 2023

iture123 Mar 19, 2023

zhilinli123 Mar 20, 2023

Hisoka-X Mar 22, 2023

zhilinli123 Mar 22, 2023 •

edited

Loading

Hisoka-X Mar 22, 2023

zhilinli123 Mar 22, 2023

zhilinli123 Mar 22, 2023

zhilinli123 commented Mar 22, 2023

hailin0 Mar 27, 2023

zhilinli123 Mar 27, 2023

hailin0 Mar 27, 2023

hailin0 Mar 27, 2023

zhilinli123 Mar 27, 2023

Hisoka-X left a comment

zhilinli123 commented May 5, 2023

TyrantLucifer commented May 5, 2023

		if (requestEsList.size() >= maxBatchSize) {
		log.info("Batch write completion row :" + requestEsList.size());

		@@ -51,6 +54,8 @@

		private int maxRetryCount = MAX_RETRY_COUNT.defaultValue();

		private int batchIntervalMs = BATCH_INTERVAL_MS.defaultValue();

	private int batchIntervalMs = BATCH_INTERVAL_MS.defaultValue();
	private Integer batchIntervalMs = BATCH_INTERVAL_MS.defaultValue();

[Feature][Connector-V2][Elasticsearch] Elasticsearch supports spaced write and is relatively real-time data source friendly #4347 #4349

[Feature][Connector-V2][Elasticsearch] Elasticsearch supports spaced write and is relatively real-time data source friendly #4347 #4349

Conversation

zhilinli123 commented Mar 14, 2023

Purpose of this pull request

Check list

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zhilinli123 Mar 22, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zhilinli123 commented Mar 22, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Hisoka-X left a comment

Choose a reason for hiding this comment

zhilinli123 commented May 5, 2023

TyrantLucifer commented May 5, 2023

zhilinli123 Mar 22, 2023 •

edited

Loading