[Feature][ConnectorV2]add file excel sink #2585

Bingz2 · 2022-08-31T15:32:43Z

Purpose of this pull request

add [File]excel sink
#1946

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

CalvinKirs · 2022-09-01T00:47:53Z

@TyrantLucifer Hi, PTAL

hailin0 · 2022-09-01T02:28:41Z

...onnector-v2-e2e/src/test/java/org/apache/seatunnel/e2e/spark/v2/file/FakeSourceToFileIT.java

+    @Test
+    public void testFakeSourceToHdfsFileExcel() throws IOException, InterruptedException {
+        Container.ExecResult execResult = executeSeaTunnelSparkJob("/file/fakesource_to_hdfs_excel.conf");
+        Assertions.assertEquals(0, execResult.getExitCode());


validate sink data row & datatypes？

hailin0 · 2022-09-01T02:29:10Z

...l-e2e/seatunnel-spark-connector-v2-e2e/src/test/resources/file/fakesource_to_hdfs_excel.conf

+        name = "string"
+        age = "int"


Test all datatypes？

test string,boolean,tinyint,smallint,int,bigint,float,double,null is ok

hailin0 · 2022-09-01T02:31:39Z

...onnector-v2-e2e/src/test/java/org/apache/seatunnel/e2e/flink/v2/file/FakeSourceToFileIT.java

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


uncommited fakesource_to_local_excel.conf、fakesource_to_hdfs_excel.conf?

TyrantLucifer · 2022-09-02T01:31:28Z

...e/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/util/ExcelGenerator.java

+    }
+
+    public void writeData(SeaTunnelRow seaTunnelRow) {
+        Sheet st = wb.getSheet("Sheet1");


How about treated st as a class attribute, when a SeaTunnelRow needs be consumed createSheet will be invoked. Of course, I have no detailed source code understanding, in each write whether to obtain a sheet object needs a double check.

oh ,Let me modify that

TyrantLucifer · 2022-09-02T05:58:11Z

You e2e test cases also have some problems, please check.

Bingz2 · 2022-09-02T13:05:07Z

You e2e test cases also have some problems, please check.

If I add this code, Excel sink Write's finishAndCloseWriteFile method will be called twice when executed with Flink, resulting in an error!

However, if you remove this code and run it with Spark, Text sink Write is executed instead of Excel sink Write

TyrantLucifer · 2022-09-03T08:26:11Z

You e2e test cases also have some problems, please check.

If I add this code, Excel sink Write's finishAndCloseWriteFile method will be called twice when executed with Flink, resulting in an error!

However, if you remove this code and run it with Spark, Text sink Write is executed instead of Excel sink Write

Can you provide the detailed call stack information?

Bingz2 · 2022-09-03T09:12:28Z

You e2e test cases also have some problems, please check.

If I add this code, Excel sink Write's finishAndCloseWriteFile method will be called twice when executed with Flink, resulting in an error!
However, if you remove this code and run it with Spark, Text sink Write is executed instead of Excel sink Write

Can you provide the detailed call stack information?

SparkDataWriter's commit() method clears commitInfo after execution, However, the prepareCommit of FlinkSinkWrite does not clear commitInfo, so it writes ExcelWorkBook when it calls the close() method of FlinkSlinkWrite, but it does the WorkBook when it executes prepareCommit It has been closed.
I don't understand why SparkDataWriter and FlinkSinkWrite have different commit logic. Do I need to determine whether the workbook is close in finishAndCloseWriteFile?

Hisoka-X · 2022-09-05T08:15:27Z

You e2e test cases also have some problems, please check.

If I add this code, Excel sink Write's finishAndCloseWriteFile method will be called twice when executed with Flink, resulting in an error!
However, if you remove this code and run it with Spark, Text sink Write is executed instead of Excel sink Write

Can you provide the detailed call stack information?

SparkDataWriter's commit() method clears commitInfo after execution, However, the prepareCommit of FlinkSinkWrite does not clear commitInfo, so it writes ExcelWorkBook when it calls the close() method of FlinkSlinkWrite, but it does the WorkBook when it executes prepareCommit It has been closed. I don't understand why SparkDataWriter and FlinkSinkWrite have different commit logic. Do I need to determine whether the workbook is close in finishAndCloseWriteFile?

The reason already in comment: combine the prepareCommit and commit in this method.. Flink Support Committer and GlobalCommitter, but spark only support GlobalCommitter(same logic different name). So spark use commit() to run prepareCommit() and Committer.commit().

Bingz2 · 2022-09-05T13:18:11Z

You e2e test cases also have some problems, please check.

If I add this code, Excel sink Write's finishAndCloseWriteFile method will be called twice when executed with Flink, resulting in an error!
However, if you remove this code and run it with Spark, Text sink Write is executed instead of Excel sink Write

Can you provide the detailed call stack information?

SparkDataWriter's commit() method clears commitInfo after execution, However, the prepareCommit of FlinkSinkWrite does not clear commitInfo, so it writes ExcelWorkBook when it calls the close() method of FlinkSlinkWrite, but it does the WorkBook when it executes prepareCommit It has been closed. I don't understand why SparkDataWriter and FlinkSinkWrite have different commit logic. Do I need to determine whether the workbook is close in finishAndCloseWriteFile?

The reason already in comment: combine the prepareCommit and commit in this method.. Flink Support Committer and GlobalCommitter, but spark only support GlobalCommitter(same logic different name). So spark use commit() to run prepareCommit() and Committer.commit().

ok,My local E2E and compile passed, but CI/CD failed and I didn't use ES.

Hisoka-X · 2022-09-06T03:15:25Z

pom.xml

@@ -758,6 +760,19 @@
                <artifactId>aliyun-sdk-datahub</artifactId>
                <version>${datahub.version}</version>
            </dependency>
+
+            <dependency>


Hi, please move connector dependency to connector's pom.xml. Reference: #2630

Ok, they have been moved to connector's pom.xml

hailin0 · 2022-09-07T12:27:41Z

...e/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/util/ExcelGenerator.java

+            cell.setCellValue((int) value);
+            cell.setCellStyle(wholeNumberCellStyle);
+        } else if (BasicType.LONG_TYPE.equals(type) || type.getSqlType().equals(SqlType.TIMESTAMP)) {
+            cell.setCellValue((long) value);


value type is LocalDateTime?

hailin0 · 2022-09-07T12:28:14Z

...e/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/util/ExcelGenerator.java

+            cell.setCellValue(JsonUtils.toJsonString(value));
+            cell.setCellStyle(stringCellStyle);
+        } else if (type.getSqlType().equals(SqlType.DATE) || type.getSqlType().equals(SqlType.TIME)) {
+            cell.setCellValue((String) value);


value type is LocalTime or LocalDate?

hailin0 · 2022-09-07T12:30:11Z

...e/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/util/ExcelGenerator.java

+        } else if (BasicType.DOUBLE_TYPE.equals(type)) {
+            cell.setCellValue((double) value);
+            cell.setCellStyle(wholeNumberCellStyle);
+        } else if (type.getSqlType().equals(SqlType.BYTES) || type.getSqlType().equals(SqlType.ARRAY)) {


ARRAY should use json?

...l-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/file/fakesource_to_hdfs_excel.conf

...-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/file/fakesource_to_local_excel.conf

hailin0 · 2022-09-07T12:40:20Z

...l-e2e/seatunnel-spark-connector-v2-e2e/src/test/resources/file/fakesource_to_hdfs_excel.conf

@@ -0,0 +1,82 @@
+#


add testcase

https://github.com/apache/incubator-seatunnel/blob/dev/seatunnel-e2e/seatunnel-spark-connector-v2-e2e/connector-file-spark-e2e/src/test/java/org/apache/seatunnel/e2e/spark/v2/file/FakeSourceToFileIT.java

hailin0 · 2022-09-07T12:40:28Z

...-e2e/seatunnel-spark-connector-v2-e2e/src/test/resources/file/fakesource_to_local_excel.conf

@@ -0,0 +1,82 @@
+#


add testcase

https://github.com/apache/incubator-seatunnel/blob/dev/seatunnel-e2e/seatunnel-spark-connector-v2-e2e/connector-file-spark-e2e/src/test/java/org/apache/seatunnel/e2e/spark/v2/file/FakeSourceToFileIT.java

hailin0 · 2022-09-07T13:19:40Z

...e/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/util/ExcelGenerator.java

+    public ExcelGenerator(List<Integer> sinkColumnsIndexInRow, SeaTunnelRowType seaTunnelRowType) {
+        this.sinkColumnsIndexInRow = sinkColumnsIndexInRow;
+        this.seaTunnelRowType = seaTunnelRowType;
+        wb = new XSSFWorkbook();


XSSFWorkbook store data to memory, is this acceptable? you consider using SXSSFWorkbook? SXSSFWorkbook supports streaming data writing

@TyrantLucifer @Hisoka-X @CalvinKirs @EricJoy2048

reference
https://github.com/apache/poi/blob/trunk/poi-ooxml/src/main/java/org/apache/poi/xssf/streaming/SXSSFWorkbook.java
https://github.com/apache/poi/blob/trunk/poi-ooxml/src/main/java/org/apache/poi/xssf/usermodel/XSSFWorkbook.java

Agree with you.

When using SXSSFWorkbook to create a sheet,local operation is normal, but E2E will report an error.
using openjdk will report a null pointer exception because there is no font.
reference:
https://blog.csdn.net/progammer10086/article/details/107154814?spm=1001.2101.3001.6650.3&utm_medium=distribute.pc_relevant.none-task-blog-2%7Edefault%7ECTRLIST%7ERate-3-107154814-blog-116048731.t0_edu_mlt&depth_1-utm_source=distribute.pc_relevant.none-task-blog-2%7Edefault%7ECTRLIST%7ERate-3-107154814-blog-116048731.t0_edu_mlt&utm_relevant_index=6

Bingz2 · 2022-10-30T09:29:00Z

Hi, @Bingz2 thanks for your contribution, what's the news about this pr? Do you need some help?
Sorry, I'm a little busy recently. I'll finish it this weekend

Thank you very much.

An error in the CI,but I executed E2E locally with no errors,I don't know why.

EricJoy2048 · 2022-10-31T09:50:17Z

Hi, @Bingz2 thanks for your contribution, what's the news about this pr? Do you need some help?
Sorry, I'm a little busy recently. I'll finish it this weekend

Thank you very much.

An error in the CI,but I executed E2E locally with no errors,I don't know why.

Don't worry, just resolve conflicts and retry the CI, Thanks.

hailin0 · 2022-11-17T02:38:20Z

docs/en/connector-v2/sink/HdfsFile.md

@@ -86,6 +87,10 @@ The separator between columns in a row of data. Only needed by `text` and `csv`

 The separator between rows in a file. Only needed by `text` and `csv` file format.

+### max_rows_in_memory [int]


add max_rows_in_memory to options

hailin0 · 2022-11-17T02:42:31Z

...e/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/util/ExcelGenerator.java

+        if (textFileSinkConfig.getMaxRowsInMemory() > 0) {
+            wb = new SXSSFWorkbook(textFileSinkConfig.getMaxRowsInMemory());
+        } else {
+            wb = new XSSFWorkbook();


MaxRowsInMemory default value is Long.MAX_VALUE ?

You can choose from the following:

It is stated in the document that if MaxRowsInMemory is not set, all data will be buffered to memory

Set default value xxx for MaxRowsInMemory

ok, I'm going to set a default

hailin0 · 2022-11-17T02:44:27Z

...l-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/file/fakesource_to_hdfs_excel.conf

@@ -0,0 +1,70 @@
+#


move this file to https://github.com/apache/incubator-seatunnel/tree/dev/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-local-e2e/src/test/resources

testing run this file
Add this file path to here https://github.com/apache/incubator-seatunnel/blob/dev/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-local-e2e/src/test/java/org/apache/seatunnel/e2e/connector/file/local/LocalFileIT.java#L52

hailin0 · 2022-11-17T02:44:41Z

...-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/file/fakesource_to_local_excel.conf

@@ -0,0 +1,69 @@
+#


move this file to https://github.com/apache/incubator-seatunnel/tree/dev/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-local-e2e/src/test/resources

testing run this file
Add this file path to here https://github.com/apache/incubator-seatunnel/blob/dev/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-file-local-e2e/src/test/java/org/apache/seatunnel/e2e/connector/file/local/LocalFileIT.java#L52

hailin0 · 2022-11-17T02:48:37Z

...l-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/file/fakesource_to_hdfs_excel.conf

+        name = "string"
+        age = "int"


test all data type write to excel file

Suggested change

name = "string"

age = "int"

c_map = "map<string, string>"

c_array = "array<int>"

c_string = string

c_boolean = boolean

c_tinyint = tinyint

c_smallint = smallint

c_int = int

c_bigint = bigint

c_float = float

c_double = double

c_bytes = bytes

c_date = date

c_decimal = "decimal(38, 18)"

c_timestamp = timestamp

c_row = {

c_map = "map<string, string>"

c_array = "array<int>"

c_string = string

c_boolean = boolean

c_tinyint = tinyint

c_smallint = smallint

c_int = int

c_bigint = bigint

c_float = float

c_double = double

c_bytes = bytes

c_date = date

c_decimal = "decimal(38, 18)"

c_timestamp = timestamp

hailin0 · 2022-11-17T02:49:07Z

...-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/file/fakesource_to_local_excel.conf

+        name = "string"
+        age = "int"


test all data type write to excel file

Suggested change

name = "string"

age = "int"

c_map = "map<string, string>"

c_array = "array<int>"

c_string = string

c_boolean = boolean

c_tinyint = tinyint

c_smallint = smallint

c_int = int

c_bigint = bigint

c_float = float

c_double = double

c_bytes = bytes

c_date = date

c_decimal = "decimal(38, 18)"

c_timestamp = timestamp

c_row = {

c_map = "map<string, string>"

c_array = "array<int>"

c_string = string

c_boolean = boolean

c_tinyint = tinyint

c_smallint = smallint

c_int = int

c_bigint = bigint

c_float = float

c_double = double

c_bytes = bytes

c_date = date

c_decimal = "decimal(38, 18)"

c_timestamp = timestamp

hailin0 · 2022-11-17T02:49:40Z

...-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/file/fakesource_to_local_excel.conf

+    is_partition_field_write_in_file=true
+    file_name_expression="${transactionId}_${now}"
+    file_format="excel"
+    sink_columns=["name","age"]


remove

Suggested change

sink_columns=["name","age"]

hailin0 · 2022-11-17T02:50:21Z

...-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/file/fakesource_to_local_excel.conf

+transform {
+  sql {
+    sql = "select name,age from fake"
+  }
+
+  # If you would like to get more information about how to configure seatunnel and see full list of transform plugins,
+  # please go to https://seatunnel.apache.org/docs/flink/configuration/transform-plugins/Sql
+}


remove

Suggested change

transform {

sql {

sql = "select name,age from fake"

}

# If you would like to get more information about how to configure seatunnel and see full list of transform plugins,

# please go to https://seatunnel.apache.org/docs/flink/configuration/transform-plugins/Sql

}

hailin0 · 2022-11-17T02:50:46Z

...l-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/file/fakesource_to_hdfs_excel.conf

+    is_partition_field_write_in_file=true
+    file_name_expression="${transactionId}_${now}"
+    file_format="excel"
+    sink_columns=["name","age"]


remove

Suggested change

sink_columns=["name","age"]

hailin0 · 2022-11-17T02:51:01Z

...l-e2e/seatunnel-flink-connector-v2-e2e/src/test/resources/file/fakesource_to_hdfs_excel.conf

+transform {
+  sql {
+    sql = "select name,age from fake"
+  }
+
+  # If you would like to get more information about how to configure seatunnel and see full list of transform plugins,
+  # please go to https://seatunnel.apache.org/docs/transform/sql
+}


remove

Suggested change

transform {

sql {

sql = "select name,age from fake"

}

# If you would like to get more information about how to configure seatunnel and see full list of transform plugins,

# please go to https://seatunnel.apache.org/docs/transform/sql

}

hi, I need to install fonts to use SXSSFWorkbook, how do I install them in the test container? For example, the following commandRUN apk add --update font-adobe-100dpi ttf-dejavu fontconfig

Maybe you need to write dockerfile to do that.Add the required content based on this image, then generate your own image and push it to docker hub.

MonsterChenzhuo · 2023-01-17T13:15:57Z

@Bingz2 Will this pr go ahead? If not, I would like to take over and finish it

Bingz2 · 2023-01-18T06:14:11Z

@Bingz2 Will this pr go ahead? If not, I would like to take over and finish it

ok tks

add file excel sink

049652d

hailin0 reviewed Sep 1, 2022

View reviewed changes

Bingz2 and others added 2 commits September 1, 2022 21:30

Merge branch 'apache:dev' into dev

e525451

add flink e2e conf,support all datatype

89e2b4f

TyrantLucifer reviewed Sep 2, 2022

View reviewed changes

modify sheet to member variable

0a5fd71

TyrantLucifer added the connectors-v2 label Sep 2, 2022

Bingz2 and others added 2 commits September 4, 2022 13:34

Merge branch 'apache:dev' into dev

b64e547

fix flink e2e

1265c23

Hisoka-X added the Waiting for users feedback Waiting for feedback from issue/PR author label Sep 5, 2022

Hisoka-X reviewed Sep 6, 2022

View reviewed changes

zhaobingzhi and others added 6 commits September 6, 2022 14:59

move excel dependency to file-connector

a595be9

Handling code conflicts

182f009

Handling code conflicts

4ad6160

Handling code conflicts

c218193

Merge branch 'apache:dev' into dev

c5470d4

Merge branch 'apache:dev' into dev

4973b36

hailin0 reviewed Sep 7, 2022

View reviewed changes

Bingz2 and others added 4 commits September 9, 2022 15:25

Merge branch 'apache:dev' into dev

e6178bb

Merge branch 'apache:dev' into dev

0fe3dfe

Handling code conflicts

93cbbb0

Merge branch 'apache:dev' into dev

a9e57f1

EricJoy2048 mentioned this pull request Sep 21, 2022

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

Closed

EricJoy2048 requested review from TyrantLucifer, Hisoka-X and hailin0 October 25, 2022 03:14

EricJoy2048 closed this Oct 28, 2022

EricJoy2048 reopened this Oct 28, 2022

Bingz2 added 2 commits October 29, 2022 14:16

Merge branch 'apache:dev' into dev

162ba2f

Merge branch 'apache:dev' into dev

7605250

EricJoy2048 closed this Oct 31, 2022

EricJoy2048 reopened this Oct 31, 2022

zhaobingzhi and others added 2 commits November 2, 2022 17:41

Resolving code conflicts

971bc78

Merge branch 'apache:dev' into dev

00134e4

EricJoy2048 closed this Nov 3, 2022

EricJoy2048 reopened this Nov 3, 2022

EricJoy2048 closed this Nov 4, 2022

EricJoy2048 reopened this Nov 4, 2022

EricJoy2048 closed this Nov 4, 2022

EricJoy2048 reopened this Nov 4, 2022

zhaobingzhi and others added 2 commits November 11, 2022 17:24

Resolving code conflicts

874af27

Merge branch 'apache:dev' into dev

4f8d572

hailin0 reviewed Nov 17, 2022

View reviewed changes

zhaobingzhi and others added 3 commits November 18, 2022 10:21

Resolving code conflicts

5754104

Merge branch 'apache:dev' into dev

78f8b2d

Merge branch 'apache:dev' into dev

9870a07

insist777 mentioned this pull request Jan 16, 2023

[Feature][Statistics] Connector pr not yet merged #3960

Closed

3 tasks

TyrantLucifer assigned MonsterChenzhuo Jan 18, 2023

Bingz2 closed this Jan 18, 2023

		@@ -86,6 +87,10 @@ The separator between columns in a row of data. Only needed by `text` and `csv`

		The separator between rows in a file. Only needed by `text` and `csv` file format.

		### max_rows_in_memory [int]

-        name = "string"
-        age = "int"
+        c_map = "map<string, string>"
+        c_array = "array<int>"
+        c_string = string
+        c_boolean = boolean
+        c_tinyint = tinyint
+        c_smallint = smallint
+        c_int = int
+        c_bigint = bigint
+        c_float = float
+        c_double = double
+        c_bytes = bytes
+        c_date = date
+        c_decimal = "decimal(38, 18)"
+        c_timestamp = timestamp
+        c_row = {
+          c_map = "map<string, string>"
+          c_array = "array<int>"
+          c_string = string
+          c_boolean = boolean
+          c_tinyint = tinyint
+          c_smallint = smallint
+          c_int = int
+          c_bigint = bigint
+          c_float = float
+          c_double = double
+          c_bytes = bytes
+          c_date = date
+          c_decimal = "decimal(38, 18)"
+          c_timestamp = timestamp

[Feature][ConnectorV2]add file excel sink #2585

[Feature][ConnectorV2]add file excel sink #2585

Conversation

Bingz2 commented Aug 31, 2022

Purpose of this pull request

Check list

CalvinKirs commented Sep 1, 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

TyrantLucifer commented Sep 2, 2022

Bingz2 commented Sep 2, 2022

TyrantLucifer commented Sep 3, 2022

Bingz2 commented Sep 3, 2022

Hisoka-X commented Sep 5, 2022

Bingz2 commented Sep 5, 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

hailin0 Sep 7, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Bingz2 commented Oct 30, 2022

EricJoy2048 commented Oct 31, 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

MonsterChenzhuo commented Jan 17, 2023

Bingz2 commented Jan 18, 2023

hailin0 Sep 7, 2022 •

edited

Loading