[Feature][ConnectorV2]add file excel sink

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

@@ -40,7 +40,7 @@ Please note that, If `is_enable_transaction` is `true`, we will auto add `${tran
 
 We supported as the following file types:
 
-`text` `csv` `parquet` `orc` `json`
+`text` `csv` `parquet` `orc` `json` `excel`
 
 Please note that, The final file name will ends with the file_format's suffix, the suffix of the text file is `txt`.
 
@@ -161,3 +161,21 @@ HdfsFile {
 }
 
 ```
+
+For excel file format
+
+```bash
+
+HdfsFile {
+    path="hdfs://mycluster/tmp/hive/warehouse/test2"
+    partition_by=["age"]
+    partition_dir_expression="${k0}=${v0}"
+    is_partition_field_write_in_file=true
+    file_name_expression="${transactionId}_${now}"
+    file_format="excel"
+    sink_columns=["name","age"]
+    filename_time_format="yyyy.MM.dd"
+    is_enable_transaction=true
+}
+
+```

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

@@ -38,7 +38,7 @@ Please note that, If `is_enable_transaction` is `true`, we will auto add `${tran
 
 We supported as the following file types:
 
-`text` `csv` `parquet` `orc` `json`
+`text` `csv` `parquet` `orc` `json` `excel`
 
 Please note that, The final file name will ends with the file_format's suffix, the suffix of the text file is `txt`.
 
@@ -159,3 +159,21 @@ LocalFile {
 }
 
 ```
+
+For excel file format
+
+```bash
+
+LocalFile {
+    path="file:///tmp/hive/warehouse/test2"
+    partition_by=["age"]
+    partition_dir_expression="${k0}=${v0}"
+    is_partition_field_write_in_file=true
+    file_name_expression="${transactionId}_${now}"
+    file_format="excel"
+    sink_columns=["name","age"]
+    filename_time_format="yyyy.MM.dd"
+    is_enable_transaction=true
+}
+
+```

pom.xml

@@ -233,6 +233,8 @@
         <neo4j-java-driver.version>4.4.9</neo4j-java-driver.version>
         <jedis.version>4.2.2</jedis.version>
         <datahub.version>2.19.0-public</datahub.version>
+        <poi.version>4.1.2</poi.version>
+        <poi-ooxml.version>4.1.2</poi-ooxml.version>
     </properties>
 
     <dependencyManagement>
@@ -997,6 +999,19 @@
                 <artifactId>aliyun-sdk-datahub</artifactId>
                 <version>${datahub.version}</version>
             </dependency>
+
+            <dependency>
+                <groupId>org.apache.poi</groupId>
+                <artifactId>poi</artifactId>
+                <version>${poi.version}</version>
+            </dependency>
+
+            <dependency>
+                <groupId>org.apache.poi</groupId>
+                <artifactId>poi-ooxml</artifactId>
+                <version>${poi-ooxml.version}</version>
+            </dependency>
+
         </dependencies>
 
     </dependencyManagement>

seatunnel-connectors-v2/connector-file/connector-file-base/pom.xml

@@ -93,5 +93,16 @@
             <scope>provided</scope>
         </dependency>
 
+
+        <dependency>
+            <groupId>org.apache.poi</groupId>
+            <artifactId>poi</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.poi</groupId>
+            <artifactId>poi-ooxml</artifactId>
+        </dependency>
+
     </dependencies>
 </project>

seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/config/FileFormat.java

@@ -18,11 +18,13 @@
 package org.apache.seatunnel.connectors.seatunnel.file.config;
 
 import org.apache.seatunnel.connectors.seatunnel.file.sink.config.TextFileSinkConfig;
+import org.apache.seatunnel.connectors.seatunnel.file.sink.writer.ExcelWriteStrategy;
 import org.apache.seatunnel.connectors.seatunnel.file.sink.writer.JsonWriteStrategy;
 import org.apache.seatunnel.connectors.seatunnel.file.sink.writer.OrcWriteStrategy;
 import org.apache.seatunnel.connectors.seatunnel.file.sink.writer.ParquetWriteStrategy;
 import org.apache.seatunnel.connectors.seatunnel.file.sink.writer.TextWriteStrategy;
 import org.apache.seatunnel.connectors.seatunnel.file.sink.writer.WriteStrategy;
+import org.apache.seatunnel.connectors.seatunnel.file.source.reader.ExcelReadStrategy;
 import org.apache.seatunnel.connectors.seatunnel.file.source.reader.JsonReadStrategy;
 import org.apache.seatunnel.connectors.seatunnel.file.source.reader.OrcReadStrategy;
 import org.apache.seatunnel.connectors.seatunnel.file.source.reader.ParquetReadStrategy;
@@ -87,6 +89,17 @@ public WriteStrategy getWriteStrategy(TextFileSinkConfig textFileSinkConfig) {
         public ReadStrategy getReadStrategy() {
             return new JsonReadStrategy();
         }
+    },
+    EXCEL("xlsx") {
+        @Override
+        public WriteStrategy getWriteStrategy(TextFileSinkConfig textFileSinkConfig) {
+            return new ExcelWriteStrategy(textFileSinkConfig);
+        }
+
+        @Override
+        public ReadStrategy getReadStrategy() {
+            return new ExcelReadStrategy();
+        }
     };
 
     private final String suffix;

seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/util/ExcelGenerator.java

@@ -0,0 +1,117 @@
+/*
+ * 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.connectors.seatunnel.file.sink.util;
+
+import org.apache.seatunnel.api.table.type.BasicType;
+import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+
+import org.apache.poi.ss.usermodel.Cell;
+import org.apache.poi.ss.usermodel.CellStyle;
+import org.apache.poi.ss.usermodel.CreationHelper;
+import org.apache.poi.ss.usermodel.Row;
+import org.apache.poi.ss.usermodel.Sheet;
+import org.apache.poi.ss.usermodel.Workbook;
+import org.apache.poi.xssf.usermodel.XSSFWorkbook;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.List;
+
+public class ExcelGenerator {
+    private Workbook wb;
+    private CellStyle wholeNumberCellStyle;
+    private CellStyle stringCellStyle;
+    private int row = 0;
+    private List<Integer> sinkColumnsIndexInRow;
+    private SeaTunnelRowType seaTunnelRowType;
+
+    public ExcelGenerator(List<Integer> sinkColumnsIndexInRow, SeaTunnelRowType seaTunnelRowType) {
+        this.sinkColumnsIndexInRow = sinkColumnsIndexInRow;
+        this.seaTunnelRowType = seaTunnelRowType;
+        wb = new XSSFWorkbook();
+        Sheet st = wb.createSheet("Sheet1");
+        Row row = st.createRow(this.row);
+        for (Integer i : sinkColumnsIndexInRow) {
+            String fieldName = seaTunnelRowType.getFieldName(i);
+            row.createCell(i).setCellValue(fieldName);
+        }
+
+        wholeNumberCellStyle = createStyle(wb, "General");
+        stringCellStyle = createStyle(wb, "@");
+        this.row += 1;
+    }
+
+    public void writeData(SeaTunnelRow seaTunnelRow) {
+        Sheet st = wb.getSheet("Sheet1");
+        Row excelRow = st.createRow(this.row);
+        SeaTunnelDataType<?>[] fieldTypes = seaTunnelRowType.getFieldTypes();
+        for (Integer i : sinkColumnsIndexInRow) {
+            Cell cell = excelRow.createCell(i);
+            Object value = seaTunnelRow.getField(i);
+            makeConverter(fieldTypes[i], value, cell);
+        }
+        this.row += 1;
+    }
+
+    public void flushAndCloseExcel(OutputStream output) throws IOException {
+        wb.write(output);
+        wb.close();
+    }
+
+    private void makeConverter(SeaTunnelDataType<?> type, Object value, Cell cell) {
+        if (value == null) {
+            cell.setBlank();
+        } else if (BasicType.STRING_TYPE.equals(type)) {
+            cell.setCellValue((String) value);
+            cell.setCellStyle(stringCellStyle);
+        } else if (BasicType.BOOLEAN_TYPE.equals(type)) {
+            cell.setCellValue((Boolean) value);
+            cell.setCellStyle(wholeNumberCellStyle);
+        } else if (BasicType.BYTE_TYPE.equals(type)) {
+            cell.setCellValue((byte) value);
+            cell.setCellStyle(wholeNumberCellStyle);
+        } else if (BasicType.SHORT_TYPE.equals(type)) {
+            cell.setCellValue((short) value);
+            cell.setCellStyle(wholeNumberCellStyle);
+        } else if (BasicType.INT_TYPE.equals(type)) {
+            cell.setCellValue((int) value);
+            cell.setCellStyle(wholeNumberCellStyle);
+        } else if (BasicType.LONG_TYPE.equals(type)) {
+            cell.setCellValue((long) value);
+            cell.setCellStyle(wholeNumberCellStyle);
+        } else if (BasicType.FLOAT_TYPE.equals(type)) {
+            cell.setCellValue((float) value);
+            cell.setCellStyle(wholeNumberCellStyle);
+        } else if (BasicType.DOUBLE_TYPE.equals(type)) {
+            cell.setCellValue((double) value);
+            cell.setCellStyle(wholeNumberCellStyle);
+        } else {
+            String errorMsg = String.format("[%s] type not support ", type.getSqlType());
+            throw new RuntimeException(errorMsg);
+        }
+    }
+
+    private CellStyle createStyle(Workbook wb, String format) {
+        CreationHelper creationHelper = wb.getCreationHelper();
+        CellStyle cellStyle = wb.createCellStyle();
+        cellStyle.setDataFormat(creationHelper.createDataFormat().getFormat(format));
+        return cellStyle;
+    }
+}

seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/writer/ExcelWriteStrategy.java

@@ -0,0 +1,71 @@
+/*
+ * 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.connectors.seatunnel.file.sink.writer;
+
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.connectors.seatunnel.file.sink.config.TextFileSinkConfig;
+import org.apache.seatunnel.connectors.seatunnel.file.sink.util.ExcelGenerator;
+import org.apache.seatunnel.connectors.seatunnel.file.sink.util.FileSystemUtils;
+
+import lombok.NonNull;
+import org.apache.hadoop.fs.FSDataOutputStream;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+
+public class ExcelWriteStrategy extends AbstractWriteStrategy {
+    private Map<String, ExcelGenerator> beingWrittenWriter;
+
+    public ExcelWriteStrategy(TextFileSinkConfig textFileSinkConfig) {
+        super(textFileSinkConfig);
+        this.beingWrittenWriter = new HashMap<>();
+    }
+
+    @Override
+    public void write(SeaTunnelRow seaTunnelRow) throws Exception {
+        String filePath = getOrCreateFilePathBeingWritten(seaTunnelRow);
+        ExcelGenerator excelGenerator = getOrCreateExcelGenerator(filePath);
+        excelGenerator.writeData(seaTunnelRow);
+    }
+
+    @Override
+    public void finishAndCloseFile() {
+        this.beingWrittenWriter.forEach((k, v) -> {
+            try {
+                FileSystemUtils.createFile(k);
+                FSDataOutputStream fileOutputStream = FileSystemUtils.getOutputStream(k);
+                v.flushAndCloseExcel(fileOutputStream);
+                fileOutputStream.close();
+            } catch (IOException e) {
+                log.error("can not get output file stream");
+                throw new RuntimeException(e);
+            }
+            needMoveFiles.put(k, getTargetLocation(k));
+        });
+    }
+
+    private ExcelGenerator getOrCreateExcelGenerator(@NonNull String filePath) {
+        ExcelGenerator excelGenerator = this.beingWrittenWriter.get(filePath);
+        if (excelGenerator == null) {
+            excelGenerator = new ExcelGenerator(sinkColumnsIndexInRow, seaTunnelRowType);
+            this.beingWrittenWriter.put(filePath, excelGenerator);
+        }
+        return excelGenerator;
+    }
+}

seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/ExcelReadStrategy.java

@@ -0,0 +1,36 @@
+/*
+ * 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.connectors.seatunnel.file.source.reader;
+
+import org.apache.seatunnel.api.source.Collector;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import org.apache.seatunnel.connectors.seatunnel.file.config.HadoopConf;
+import org.apache.seatunnel.connectors.seatunnel.file.exception.FilePluginException;
+
+public class ExcelReadStrategy extends AbstractReadStrategy {
+    @Override
+    public void read(String path, Collector<SeaTunnelRow> output) throws Exception {
+
+    }
+
+    @Override
+    public SeaTunnelRowType getSeaTunnelRowTypeInfo(HadoopConf hadoopConf, String path) throws FilePluginException {
+        return null;
+    }
+}