[Feature][Connector-V2][Elasticsearch] Support Elasticsearch source #2821

iture123 · 2022-09-20T15:04:03Z

)

…l into dev-20220919

…v-20220919 � Conflicts: � seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/client/EsRestClient.java

ashulin · 2022-09-27T02:30:33Z

.../seatunnel/connectors/seatunnel/elasticsearch/source/ElasticsearchSourceSplitEnumerator.java

+        indexDocsCounts = indexDocsCounts.stream().filter(x -> x.getDocsCount() != null && x.getDocsCount() > 0)
+                .sorted(Comparator.comparingLong(IndexDocsCount::getDocsCount)).collect(Collectors.toList());
+        List<ElasticsearchSourceSplit> splits = new ArrayList<>();
+        int parallelism = enumeratorContext.currentParallelism();


EnumeratorContext.currentParallelism() is Runtime Method, Use it in SplitEnumerator#run()

ashulin · 2022-09-27T02:34:06Z

...c/test/java/org/apache/seatunnel/e2e/flink/v2/elasticsearch/FakeSourceToElasticsearchIT.java

+
+    private ElasticsearchContainer container;
+
+    @SuppressWarnings({"checkstyle:MagicNumber", "checkstyle:Indentation"})


Suggested change

@SuppressWarnings({"checkstyle:MagicNumber", "checkstyle:Indentation"})

IT will automatically ignore MagicNumber, please do not ignore the indentation.

ashulin · 2022-09-27T02:36:31Z

...rg/apache/seatunnel/connectors/seatunnel/elasticsearch/source/ElasticsearchSourceReader.java

+            ScrollResult scrollResult = esRestClient.searchByScroll(sourceIndexInfo.getIndex(), sourceIndexInfo.getSource(), sourceIndexInfo.getScrollTime(), sourceIndexInfo.getScrollSize());
+            outputFromScrollResult(scrollResult, sourceIndexInfo.getSource(), output);
+            while (scrollResult.getDocs() != null && scrollResult.getDocs().size() > 0) {
+                scrollResult = esRestClient.searchWithScrollId(scrollResult.getScrollId(), sourceIndexInfo.getScrollTime());
+                outputFromScrollResult(scrollResult, sourceIndexInfo.getSource(), output);
+            }


Suggested change

ScrollResult scrollResult = esRestClient.searchByScroll(sourceIndexInfo.getIndex(), sourceIndexInfo.getSource(), sourceIndexInfo.getScrollTime(), sourceIndexInfo.getScrollSize());

outputFromScrollResult(scrollResult, sourceIndexInfo.getSource(), output);

while (scrollResult.getDocs() != null && scrollResult.getDocs().size() > 0) {

scrollResult = esRestClient.searchWithScrollId(scrollResult.getScrollId(), sourceIndexInfo.getScrollTime());

outputFromScrollResult(scrollResult, sourceIndexInfo.getSource(), output);

}

synchronized (output.getCheckpointLock()) {

ScrollResult scrollResult = esRestClient.searchByScroll(sourceIndexInfo.getIndex(), sourceIndexInfo.getSource(), sourceIndexInfo.getScrollTime(), sourceIndexInfo.getScrollSize());

outputFromScrollResult(scrollResult, sourceIndexInfo.getSource(), output);

while (scrollResult.getDocs() != null && scrollResult.getDocs().size() > 0) {

scrollResult = esRestClient.searchWithScrollId(scrollResult.getScrollId(), sourceIndexInfo.getScrollTime());

outputFromScrollResult(scrollResult, sourceIndexInfo.getSource(), output);

}

}

ashulin · 2022-09-27T02:39:04Z

...java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/dto/source/SourceIndexInfo.java

+import java.io.Serializable;
+import java.util.List;
+
+public class SourceIndexInfo implements Serializable {


Suggested change

public class SourceIndexInfo implements Serializable {

@Data

@AllArgsConstructor

public class SourceIndexInfo implements Serializable {

Please use lombok.

…' in ElasticsearchSourceReader(apache#2553)

hailin0 · 2022-09-28T14:03:57Z

docs/en/connector-v2/source/Elasticsearch.md

+Maximum number of hits to be returned with each Elasticsearch scroll request.
+
+## Examples
+simple


Add complex example

hailin0 · 2022-09-28T14:06:39Z

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

@@ -45,8 +45,7 @@ public String getPluginName() {
    }

    @Override
-    public void prepare(org.apache.seatunnel.shade.com.typesafe.config.Config pluginConfig) throws
-        PrepareFailException {
+    public void prepare(org.apache.seatunnel.shade.com.typesafe.config.Config pluginConfig) throws PrepareFailException {


rename org.apache.seatunnel.shade.com.typesafe.config.Config to Config

hailin0 · 2022-09-28T14:13:52Z

...rg/apache/seatunnel/connectors/seatunnel/elasticsearch/source/ElasticsearchSourceReader.java

+            SeaTunnelRow seaTunnelRow = new SeaTunnelRow(sourceSize);
+            for (int i = 0; i < sourceSize; i++) {
+                Object value = doc.get(source.get(i));
+                seaTunnelRow.setField(i, String.valueOf(value));


String.valueOf(value) Convert all field to string ?

you can mapping elasticsearch datatype to seatunnel datatype ?

reference
https://github.com/apache/incubator-seatunnel/tree/dev/seatunnel-connectors-v2/connector-iotdb/src/main/java/org/apache/seatunnel/connectors/seatunnel/iotdb/serialize

hailin0 · 2022-09-28T14:35:30Z

...est/java/org/apache/seatunnel/e2e/flink/v2/elasticsearch/ElasticsearchSourceToConsoleIT.java

+    @Test
+    public void testElasticsearchSourceToConsoleSink() throws IOException, InterruptedException {
+        Container.ExecResult execResult = executeSeaTunnelFlinkJob("/elasticsearch/elasticsearch_to_console.conf");
+        Assertions.assertEquals(0, execResult.getExitCode());


validate data rows & values & types when e2e execute ends

reference
https://github.com/apache/incubator-seatunnel/blob/dev/seatunnel-e2e/seatunnel-flink-connector-v2-e2e/connector-mongodb-flink-e2e/src/test/java/org/apache/seatunnel/e2e/flink/v2/mongodb/MongodbIT.java#L106

hailin0 · 2022-09-28T14:43:29Z

docs/en/connector-v2/source/Elasticsearch.md

+## Key features
+
+- [x] [batch](../../concept/connector-v2-features.md)
+- [x] [stream](../../concept/connector-v2-features.md)


support stream read?

Don't support.I will revise.

hailin0 · 2022-09-28T14:46:20Z

docs/en/connector-v2/source/Elasticsearch.md

+
+- [x] [batch](../../concept/connector-v2-features.md)
+- [x] [stream](../../concept/connector-v2-features.md)
+- [x] [exactly-once](../../concept/connector-v2-features.md)


How to restore split(exactly-once)?

reference restore split
#2917

…he#2553)

…serialze(apache#2553)

…g fix NPP(apache#2553)

…2553)

…pache#2553)

…ache#2553)

…v-20220919 � 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

…nd remove jackson-datatype-jsr310(apache#2553)

hailin0 · 2022-10-28T11:36:09Z

seatunnel-connectors-v2/connector-elasticsearch/pom.xml

        <dependency>
-            <groupId>com.fasterxml.jackson.datatype</groupId>
-            <artifactId>jackson-datatype-jsr310</artifactId>
-            <version>${jackson-datatype-jsr310.version}</version>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>seatunnel-api</artifactId>
+            <version>${project.version}</version>
+            <scope>compile</scope>
+        </dependency>


remove seatunnel-api

https://github.com/apache/incubator-seatunnel/blob/dev/seatunnel-connectors-v2/pom.xml#L64

hailin0 · 2022-10-28T11:40:32Z

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

@@ -43,43 +44,31 @@
 /**
 * ElasticsearchSinkWriter is a sink writer that will write {@link SeaTunnelRow} to Elasticsearch.
 */
-@Slf4j


why delete this？

hailin0 · 2022-10-28T11:47:24Z

...rg/apache/seatunnel/connectors/seatunnel/elasticsearch/source/ElasticsearchSourceReader.java

+        ElasticsearchSourceSplit split = splits.poll();
+        if (null != split) {
+            SourceIndexInfo sourceIndexInfo = split.getSourceIndexInfo();
+            synchronized (output.getCheckpointLock()) {
+                ScrollResult scrollResult = esRestClient.searchByScroll(sourceIndexInfo.getIndex(), sourceIndexInfo.getSource(), sourceIndexInfo.getScrollTime(), sourceIndexInfo.getScrollSize());
+                outputFromScrollResult(scrollResult, sourceIndexInfo.getSource(), output);
+                while (scrollResult.getDocs() != null && scrollResult.getDocs().size() > 0) {
+                    scrollResult = esRestClient.searchWithScrollId(scrollResult.getScrollId(), sourceIndexInfo.getScrollTime());
+                    outputFromScrollResult(scrollResult, sourceIndexInfo.getSource(), output);
+                }
+            }


move queue.poll into synchronized code block

Suggested change

ElasticsearchSourceSplit split = splits.poll();

if (null != split) {

SourceIndexInfo sourceIndexInfo = split.getSourceIndexInfo();

synchronized (output.getCheckpointLock()) {

ScrollResult scrollResult = esRestClient.searchByScroll(sourceIndexInfo.getIndex(), sourceIndexInfo.getSource(), sourceIndexInfo.getScrollTime(), sourceIndexInfo.getScrollSize());

outputFromScrollResult(scrollResult, sourceIndexInfo.getSource(), output);

while (scrollResult.getDocs() != null && scrollResult.getDocs().size() > 0) {

scrollResult = esRestClient.searchWithScrollId(scrollResult.getScrollId(), sourceIndexInfo.getScrollTime());

outputFromScrollResult(scrollResult, sourceIndexInfo.getSource(), output);

}

}

if (!splits.isEmpty()) {

synchronized (output.getCheckpointLock()) {

ElasticsearchSourceSplit split = splits.poll();

SourceIndexInfo sourceIndexInfo = split.getSourceIndexInfo();

ScrollResult scrollResult = esRestClient.searchByScroll(sourceIndexInfo.getIndex(), sourceIndexInfo.getSource(), sourceIndexInfo.getScrollTime(), sourceIndexInfo.getScrollSize());

outputFromScrollResult(scrollResult, sourceIndexInfo.getSource(), output);

while (scrollResult.getDocs() != null && scrollResult.getDocs().size() > 0) {

scrollResult = esRestClient.searchWithScrollId(scrollResult.getScrollId(), sourceIndexInfo.getScrollTime());

outputFromScrollResult(scrollResult, sourceIndexInfo.getSource(), output);

}

}

hailin0 · 2022-10-28T11:49:13Z

...org/apache/seatunnel/connectors/seatunnel/elasticsearch/source/ElasticsearchSourceSplit.java

+    public ElasticsearchSourceSplit(String splitId, SourceIndexInfo sourceIndexInfo) {
+        this.splitId = splitId;
+        this.sourceIndexInfo = sourceIndexInfo;
+    }
+
+    public SourceIndexInfo getSourceIndexInfo() {
+        return sourceIndexInfo;
+    }
+
+    @Override
+    public String splitId() {
+        return splitId;
+    }


Use lombok annotation

hailin0 · 2022-10-28T11:55:15Z

.../seatunnel/connectors/seatunnel/elasticsearch/source/ElasticsearchSourceSplitEnumerator.java

+    public void addSplitsBack(List<ElasticsearchSourceSplit> splits, int subtaskId) {
+
+    }


Suggested change

public void addSplitsBack(List<ElasticsearchSourceSplit> splits, int subtaskId) {

}

public void addSplitsBack(List<ElasticsearchSourceSplit> splits, int subtaskId) {

if (!splits.isEmpty()) {

addPendingSplit(splits);

assignSplit(Collections.singletonList(subtaskId));

}

}

hailin0 · 2022-10-28T11:58:58Z

...flink-e2e/src/test/java/org/apache/seatunnel/e2e/flink/v2/elasticsearch/ElasticsearchIT.java

+    private static final Logger LOGGER = LoggerFactory.getLogger(ElasticsearchIT.class);
+
+    @BeforeEach
+    public void startMongoContainer() throws Exception {


Suggested change

public void startMongoContainer() throws Exception {

public void startElasticsearchContainer() throws Exception {

hailin0 · 2022-10-28T12:00:30Z

...flink-e2e/src/test/java/org/apache/seatunnel/e2e/flink/v2/elasticsearch/ElasticsearchIT.java

+    }
+
+    private List<String> generateTestDataSet() throws JsonProcessingException, UnknownHostException {
+        String[] fiels = new String[]{


Suggested change

String[] fiels = new String[]{

String[] fields = new String[]{

hailin0 · 2022-10-28T12:03:32Z

...spark-e2e/src/test/java/org/apache/seatunnel/e2e/spark/v2/elasticsearch/ElasticsearchIT.java

+    private static final Logger LOGGER = LoggerFactory.getLogger(ElasticsearchIT.class);
+
+    @BeforeEach
+    public void startMongoContainer() throws Exception {


Suggested change

public void startMongoContainer() throws Exception {

public void startElasticsearchContainer() throws Exception {

hailin0 · 2022-10-28T12:03:59Z

...spark-e2e/src/test/java/org/apache/seatunnel/e2e/spark/v2/elasticsearch/ElasticsearchIT.java

+        container = new ElasticsearchContainer(DockerImageName.parse("elasticsearch:6.8.23").asCompatibleSubstituteFor("docker.elastic.co/elasticsearch/elasticsearch"))
+                .withNetwork(NETWORK)
+                .withNetworkAliases("elasticsearch")
+                .withLogConsumer(new Slf4jLogConsumer(LOGGER));


Suggested change

.withLogConsumer(new Slf4jLogConsumer(LOGGER));

.withLogConsumer(new Slf4jLogConsumer(DockerLoggerFactory.getLogger("elasticsearch:6.8.23")));

hailin0 · 2022-10-28T12:04:27Z

...spark-e2e/src/test/java/org/apache/seatunnel/e2e/spark/v2/elasticsearch/ElasticsearchIT.java

+    }
+
+    private List<String> generateTestDataSet() throws JsonProcessingException, UnknownHostException {
+        String[] fiels = new String[]{


Suggested change

String[] fiels = new String[]{

String[] fields = new String[]{

All your suggestions is improved.Thinks.

…nnel-api

@slf4j

…:use @slf4j

…:throw Exception when bulk elasticsearch error

…tEnumerator

…mbok

hailin0

LGTM

Hisoka-X

LGTM

EricJoy2048 · 2022-10-28T11:24:31Z

docs/en/connector-v2/source/Elasticsearch.md

+        }
+    }
+}
+```


Please add Change log reference https://github.com/apache/incubator-seatunnel/blob/dev/docs/en/connector-v2/source/SftpFile.md

EricJoy2048 · 2022-11-04T02:01:18Z

I will merge this pr and please add I will add a pr to change log reference https://github.com/apache/incubator-seatunnel/blob/dev/docs/en/connector-v2/source/SftpFile.md

iture123 added 2 commits September 19, 2022 22:32

[Feature][Connector-V2]improve elasticseach connector directory struc…

a96b361

…ture(apache#2553)

[Feature][Connector-V2]new connector of Elasticsearch source(apache#2553

c7050d7

)

EricJoy2048 mentioned this pull request Sep 21, 2022

[Umbrella][Connector] New SeaTunnel API Connectors #1946

Closed

iture123 added 5 commits September 22, 2022 22:33

[Feature][Connector-V2]improve elasticseach connector directory struc…

66f9ad9

…ture(apache#2553)

[Feature][Connector-V2]new connector of Elasticsearch source(apache#2553

b6688c2

)

Merge branch 'dev-20220919' of github.com:iture123/incubator-seatunne…

a8f96fe

…l into dev-20220919

Merge branch 'dev' of github.com:iture123/incubator-seatunnel into de…

3e4538e

…v-20220919 � Conflicts: � seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/client/EsRestClient.java

[Feature][Connector-V2]resolve conflict(apache#2553)

c878685

ashulin reviewed Sep 27, 2022

View reviewed changes

EricJoy2048 mentioned this pull request Sep 27, 2022

[Feature][Connector-V2] new connecotor of Elasticsearch source(#2553) #2554

Closed

iture123 added 3 commits September 27, 2022 23:15

[Feature][Connector-V2]remove checkstyle:MagicNumber(apache#2553)

5cba46f

[Feature][Connector-V2]use lombok to simple code(apache#2553)

83e288b

[Feature][Connector-V2]add 'synchronized (output.getCheckpointLock())…

c70f2af

…' in ElasticsearchSourceReader(apache#2553)

hailin0 reviewed Sep 28, 2022

View reviewed changes

ashulin changed the title ~~[Feature][Connector-V2]new connecotor of Elasticsearch source~~ [Feature][Connector-V2][Elasticsearch] Support Elasticsearch source Sep 28, 2022

hailin0 reviewed Sep 29, 2022

View reviewed changes

EricJoy2048 mentioned this pull request Oct 8, 2022

[Umbrella][Connector-V2] SeaTunnel Connector V2 Tracking Table #3018

Open

3 tasks

iture123 added 9 commits October 15, 2022 21:38

[Feature][Connector-V2][Elasticsearce]simplify Config.class path(apac…

5a02fc9

…he#2553)

[Feature][Connector-V2][Elasticsearce]improve elasticsearch record de…

ad71a2a

…serialze(apache#2553)

[Feature][Connector-V2][Elasticsearce]EsRestClient#getFieldTypeMappin…

d39c5cb

…g fix NPP(apache#2553)

[Feature][Connector-V2][Elasticsearce]improve restore split(apache#2553)

ed2ba50

[Feature][Connector-V2][Elasticsearce]improve Elasticsearc.md(apache#…

a184fac

…2553)

[Feature][Connector-V2][Elasticsearce]fix NPP when ES version <= 6.x(a…

dbb44ad

…pache#2553)

[Feature][Connector-V2][Elasticsearce]improve elasticsearch source(ap…

62f299f

…ache#2553)

[Feature][Connector-V2][Elasticsearce]fix jackson conflict in spark a…

4a3b560

…nd remove jackson-datatype-jsr310(apache#2553)

hailin0 reviewed Oct 28, 2022

View reviewed changes

iture123 added 3 commits October 30, 2022 13:12

[Feature][Connector-V2][Elasticsearce]pom.xml remove dependency seatu…

388226b

…nnel-api

[Feature][Connector-V2][Elasticsearce]improve ElasticsearchSinkWriter…

d53b879

…:use @slf4j

[Feature][Connector-V2][Elasticsearce]improve ElasticsearchSinkWriter…

b51b058

…:throw Exception when bulk elasticsearch error

iture123 added 4 commits October 31, 2022 21:46

[Feature][Connector-V2][Elasticsearce]improve ElasticsearchIT

b6c318d

[Feature][Connector-V2][Elasticsearce]improve ElasticsearchSourceReader

6eb559f

[Feature][Connector-V2][Elasticsearce]improve ElasticsearchSourceSpli…

0ccdca0

…tEnumerator

[Feature][Connector-V2][Elasticsearce]ElasticsearchSourceSplit use lo…

4232854

…mbok

hailin0 approved these changes Nov 1, 2022

View reviewed changes

Hisoka-X approved these changes Nov 2, 2022

View reviewed changes

EricJoy2048 approved these changes Nov 4, 2022

View reviewed changes

EricJoy2048 merged commit ded5481 into apache:dev Nov 4, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Feature][Connector-V2][Elasticsearch] Support Elasticsearch source #2821

[Feature][Connector-V2][Elasticsearch] Support Elasticsearch source #2821

iture123 commented Sep 20, 2022

ashulin Sep 27, 2022

ashulin Sep 27, 2022

ashulin Sep 27, 2022

iture123 Oct 2, 2022

ashulin Sep 27, 2022

hailin0 Sep 28, 2022

hailin0 Sep 28, 2022

hailin0 Sep 28, 2022

iture123 Oct 2, 2022

hailin0 Sep 28, 2022

hailin0 Sep 28, 2022

iture123 Oct 2, 2022

hailin0 Sep 28, 2022

hailin0 Oct 28, 2022

hailin0 Oct 28, 2022

hailin0 Oct 28, 2022 •

edited

Loading

hailin0 Oct 28, 2022

hailin0 Oct 28, 2022

hailin0 Oct 28, 2022

hailin0 Oct 28, 2022

hailin0 Oct 28, 2022

hailin0 Oct 28, 2022

hailin0 Oct 28, 2022

iture123 Oct 31, 2022

hailin0 left a comment

Hisoka-X left a comment

EricJoy2048 Oct 28, 2022

EricJoy2048 commented Nov 4, 2022


		private ElasticsearchContainer container;

		@SuppressWarnings({"checkstyle:MagicNumber", "checkstyle:Indentation"})

		public void addSplitsBack(List<ElasticsearchSourceSplit> splits, int subtaskId) {

		}

-    public void addSplitsBack(List<ElasticsearchSourceSplit> splits, int subtaskId) {
-    }
+    public void addSplitsBack(List<ElasticsearchSourceSplit> splits, int subtaskId) {
+        if (!splits.isEmpty()) {
+            addPendingSplit(splits);
+            assignSplit(Collections.singletonList(subtaskId));
+        }
+    }

	public void startMongoContainer() throws Exception {
	public void startElasticsearchContainer() throws Exception {

	String[] fiels = new String[]{
	String[] fields = new String[]{

	.withLogConsumer(new Slf4jLogConsumer(LOGGER));
	.withLogConsumer(new Slf4jLogConsumer(DockerLoggerFactory.getLogger("elasticsearch:6.8.23")));

[Feature][Connector-V2][Elasticsearch] Support Elasticsearch source #2821

[Feature][Connector-V2][Elasticsearch] Support Elasticsearch source #2821

Conversation

iture123 commented Sep 20, 2022

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hailin0 Oct 28, 2022 • 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

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

hailin0 left a comment

Choose a reason for hiding this comment

Hisoka-X left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

EricJoy2048 commented Nov 4, 2022

hailin0 Oct 28, 2022 •

edited

Loading