Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[Feature][Connectors] LocalFile Support reading gz #8025

Merged
merged 29 commits into from
Nov 23, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
29 commits
Select commit Hold shift + click to select a range
2a54ba3
Update ArchiveCompressFormat.java
zhdech Nov 12, 2024
bff405a
Update AbstractReadStrategy.java
zhdech Nov 12, 2024
d827c58
Update LocalFile.md
zhdech Nov 12, 2024
dae992c
solve https://github.com/apache/seatunnel/issues/8019
zhdech Nov 12, 2024
acadf05
Merge branch 'apache:dev' into dev
zhdech Nov 13, 2024
f03d2f0
Merge branch 'apache:dev' into dev
zhdech Nov 13, 2024
4205770
Merge branch 'apache:dev' into dev
zhdech Nov 14, 2024
a2d1a69
Merge branch 'apache:dev' into dev
zhdech Nov 14, 2024
0550b6c
solve https://github.com/apache/seatunnel/issues/8019
zhdech Nov 14, 2024
98a69f0
solve https://github.com/apache/seatunnel/issues/8019
zhdech Nov 14, 2024
434ed46
Merge branch 'apache:dev' into dev
zhdech Nov 14, 2024
f7ea9f9
修复包引入问题
zhdech Nov 15, 2024
c39f4b9
修复包引入问题
zhdech Nov 15, 2024
07e83a9
修复样式
zhdech Nov 15, 2024
4f19768
e2e修复文件找不到的bug
zhdech Nov 15, 2024
a4e53d7
e2e修复文件找不到的bug
zhdech Nov 15, 2024
2e7edf3
Merge remote-tracking branch 'origin/dev' into dev
zhdech Nov 15, 2024
ac5f088
e2e修复文件找不到的bug
zhdech Nov 15, 2024
2fb9736
Merge remote-tracking branch 'origin/dev' into dev
zhdech Nov 15, 2024
74e0745
Merge branch 'apache:dev' into dev
zhdech Nov 15, 2024
bf85f19
Merge branch 'apache:dev' into dev
zhdech Nov 16, 2024
03e7d38
e2e修复问题
zhdech Nov 17, 2024
cb20fbf
Merge branch 'apache:dev' into dev
zhdech Nov 17, 2024
fbbb8dc
调整容器中文件的路径。
zhdech Nov 18, 2024
9821d0b
Merge branch 'apache:dev' into dev
zhdech Nov 19, 2024
37ae65e
Remove Excel
zhdech Nov 19, 2024
480055b
format
zhdech Nov 19, 2024
a054878
Merge branch 'apache:dev' into dev
zhdech Nov 20, 2024
f4f87ed
Remove support for Excel in gz compression
zhdech Nov 20, 2024
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions docs/en/connector-v2/source/CosFile.md
Original file line number Diff line number Diff line change
Expand Up @@ -343,6 +343,7 @@ The compress codec of archive files and the details that supported as the follow
| ZIP | txt,json,excel,xml | .zip |
| TAR | txt,json,excel,xml | .tar |
| TAR_GZ | txt,json,excel,xml | .tar.gz |
| GZ | txt,json,xml | .gz |
| NONE | all | .* |

### encoding [string]
Expand Down
1 change: 1 addition & 0 deletions docs/en/connector-v2/source/FtpFile.md
Original file line number Diff line number Diff line change
Expand Up @@ -328,6 +328,7 @@ The compress codec of archive files and the details that supported as the follow
| ZIP | txt,json,excel,xml | .zip |
| TAR | txt,json,excel,xml | .tar |
| TAR_GZ | txt,json,excel,xml | .tar.gz |
| GZ | txt,json,xml | .gz |
| NONE | all | .* |

### encoding [string]
Expand Down
1 change: 1 addition & 0 deletions docs/en/connector-v2/source/HdfsFile.md
Original file line number Diff line number Diff line change
Expand Up @@ -144,6 +144,7 @@ The compress codec of archive files and the details that supported as the follow
| ZIP | txt,json,excel,xml | .zip |
| TAR | txt,json,excel,xml | .tar |
| TAR_GZ | txt,json,excel,xml | .tar.gz |
| GZ | txt,json,xml | .gz |
| NONE | all | .* |

### encoding [string]
Expand Down
3 changes: 3 additions & 0 deletions docs/en/connector-v2/source/LocalFile.md
Original file line number Diff line number Diff line change
Expand Up @@ -322,6 +322,7 @@ The compress codec of archive files and the details that supported as the follow
| ZIP | txt,json,excel,xml | .zip |
| TAR | txt,json,excel,xml | .tar |
| TAR_GZ | txt,json,excel,xml | .tar.gz |
| GZ | txt,json,xml | .gz |
| NONE | all | .* |

### encoding [string]
Expand Down Expand Up @@ -490,4 +491,6 @@ sink {
- [BugFix] Fix the bug of incorrect path in windows environment ([2980](https://github.com/apache/seatunnel/pull/2980))
- [Improve] Support extract partition from SeaTunnelRow fields ([3085](https://github.com/apache/seatunnel/pull/3085))
- [Improve] Support parse field from file path ([2985](https://github.com/apache/seatunnel/pull/2985))
### 2.3.9-beta 2024-11-12
- [Improve] Support parse field from file path ([8019](https://github.com/apache/seatunnel/issues/8019))

1 change: 1 addition & 0 deletions docs/en/connector-v2/source/OssJindoFile.md
Original file line number Diff line number Diff line change
Expand Up @@ -335,6 +335,7 @@ The compress codec of archive files and the details that supported as the follow
| ZIP | txt,json,excel,xml | .zip |
| TAR | txt,json,excel,xml | .tar |
| TAR_GZ | txt,json,excel,xml | .tar.gz |
| GZ | txt,json,xml | .gz |
| NONE | all | .* |

### encoding [string]
Expand Down
1 change: 1 addition & 0 deletions docs/en/connector-v2/source/S3File.md
Original file line number Diff line number Diff line change
Expand Up @@ -299,6 +299,7 @@ The compress codec of archive files and the details that supported as the follow
| ZIP | txt,json,excel,xml | .zip |
| TAR | txt,json,excel,xml | .tar |
| TAR_GZ | txt,json,excel,xml | .tar.gz |
| GZ | txt,json,xml | .gz |
| NONE | all | .* |

### encoding [string]
Expand Down
13 changes: 7 additions & 6 deletions docs/en/connector-v2/source/SftpFile.md
Original file line number Diff line number Diff line change
Expand Up @@ -235,11 +235,12 @@ The compress codec of files and the details that supported as the following show
The compress codec of archive files and the details that supported as the following shown:

| archive_compress_codec | file_format | archive_compress_suffix |
|------------------------|--------------------|-------------------------|
| ZIP | txt,json,excel,xml | .zip |
| TAR | txt,json,excel,xml | .tar |
| TAR_GZ | txt,json,excel,xml | .tar.gz |
| NONE | all | .* |
|--------------------|--------------------|---------------------|
| ZIP | txt,json,excel,xml | .zip |
| TAR | txt,json,excel,xml | .tar |
| TAR_GZ | txt,json,excel,xml | .tar.gz |
| GZ | txt,json,xml | .gz |
| NONE | all | .* |

### encoding [string]

Expand Down Expand Up @@ -384,4 +385,4 @@ sink {
Console {
}
}
```
```
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@ public enum ArchiveCompressFormat {
ZIP(".zip"),
TAR(".tar"),
TAR_GZ(".tar.gz"),
GZ(".gz"),
;
private final String archiveCompressCodec;

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -238,6 +238,11 @@ protected void resolveArchiveCompressedInputStream(
}
}
break;
case GZ:
GzipCompressorInputStream gzipIn =
new GzipCompressorInputStream(hadoopFileSystemProxy.getInputStream(path));
readProcess(path, tableId, output, copyInputStream(gzipIn), partitionsMap, path);
break;
case NONE:
readProcess(
path,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -50,6 +50,8 @@

import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.nio.charset.StandardCharsets;
Expand Down Expand Up @@ -149,6 +151,13 @@ public class LocalFileIT extends TestSuiteBase {
"/seatunnel/read/tar_gz/txt/multifile/multiTarGz.tar.gz",
container);

Path txtGz =
convertToGzFile(
Lists.newArrayList(ContainerUtil.getResourcesFile("/text/e2e.txt")),
"e2e-txt-gz");
ContainerUtil.copyFileIntoContainers(
txtGz, "/seatunnel/read/gz/txt/single/e2e-txt-gz.gz", container);

Path jsonZip =
convertToZipFile(
Lists.newArrayList(
Expand All @@ -168,6 +177,14 @@ public class LocalFileIT extends TestSuiteBase {
"/seatunnel/read/zip/json/multifile/multiJson.zip",
container);

Path jsonGz =
convertToGzFile(
Lists.newArrayList(
ContainerUtil.getResourcesFile("/json/e2e.json")),
"e2e-json-gz");
ContainerUtil.copyFileIntoContainers(
jsonGz, "/seatunnel/read/gz/json/single/e2e-json-gz.gz", container);

ContainerUtil.copyFileIntoContainers(
"/text/e2e_gbk.txt",
"/seatunnel/read/encoding/text/e2e_gbk.txt",
Expand All @@ -193,6 +210,13 @@ public class LocalFileIT extends TestSuiteBase {
ContainerUtil.copyFileIntoContainers(
xmlZip, "/seatunnel/read/zip/xml/single/e2e-xml.zip", container);

Path xmlGz =
convertToGzFile(
Lists.newArrayList(ContainerUtil.getResourcesFile("/xml/e2e.xml")),
"e2e-xml-gz");
ContainerUtil.copyFileIntoContainers(
xmlGz, "/seatunnel/read/gz/xml/single/e2e-xml-gz.gz", container);

Path txtLzo = convertToLzoFile(ContainerUtil.getResourcesFile("/text/e2e.txt"));
ContainerUtil.copyFileIntoContainers(
txtLzo, "/seatunnel/read/lzo_text/e2e.txt", container);
Expand Down Expand Up @@ -313,6 +337,7 @@ public void testLocalFileReadAndWrite(TestContainer container)
/** Compressed file test */
// test read single local text file with zip compression
helper.execute("/text/local_file_zip_text_to_assert.conf");
helper.execute("/text/local_file_gz_text_to_assert.conf");
// test read multi local text file with zip compression
helper.execute("/text/local_file_multi_zip_text_to_assert.conf");
// test read single local text file with tar compression
Expand All @@ -325,10 +350,12 @@ public void testLocalFileReadAndWrite(TestContainer container)
helper.execute("/text/local_file_multi_tar_gz_text_to_assert.conf");
// test read single local json file with zip compression
helper.execute("/json/local_file_json_zip_to_assert.conf");
helper.execute("/json/local_file_json_gz_to_assert.conf");
// test read multi local json file with zip compression
helper.execute("/json/local_file_json_multi_zip_to_assert.conf");
// test read single local xml file with zip compression
helper.execute("/xml/local_file_zip_xml_to_assert.conf");
helper.execute("/xml/local_file_gz_xml_to_assert.conf");
// test read single local excel file with zip compression
helper.execute("/excel/local_excel_zip_to_assert.conf");
// test read multi local excel file with zip compression
Expand Down Expand Up @@ -551,4 +578,29 @@ public FileVisitResult visitFile(

return tarGzFilePath;
}

public Path convertToGzFile(List<File> files, String name) throws IOException {
if (files == null || files.isEmpty()) {
throw new IllegalArgumentException("File list is empty or invalid");
}

File firstFile = files.get(0);
Path gzFilePath = Paths.get(firstFile.getParent(), String.format("%s.gz", name));

try (FileInputStream fis = new FileInputStream(firstFile);
FileOutputStream fos = new FileOutputStream(gzFilePath.toFile());
GZIPOutputStream gzos = new GZIPOutputStream(fos)) {

byte[] buffer = new byte[2048];
int length;

while ((length = fis.read(buffer)) > 0) {
gzos.write(buffer, 0, length);
}
gzos.finish();
} catch (IOException e) {
e.printStackTrace();
}
return gzFilePath;
}
}
Original file line number Diff line number Diff line change
@@ -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.
#

env {
parallelism = 1
job.mode = "BATCH"

# You can set spark configuration here
spark.app.name = "SeaTunnel"
spark.executor.instances = 2
spark.executor.cores = 1
spark.executor.memory = "1g"
spark.master = local
}

source {
LocalFile {
path = "/seatunnel/read/gz/json/single/e2e-json-gz.gz"
file_format_type = "json"
archive_compress_codec = "gz"
schema = {
fields {
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
}
}
}
result_table_name = "fake"
}
}

sink {
Assert {
rules {
row_rules = [
{
rule_type = MAX_ROW
rule_value = 5
},
{
rule_type = MIN_ROW
rule_value = 5
}
],
field_rules = [
{
field_name = c_string
field_type = string
field_value = [
{
rule_type = NOT_NULL
}
]
},
{
field_name = c_boolean
field_type = boolean
field_value = [
{
rule_type = NOT_NULL
}
]
},
{
field_name = c_double
field_type = double
field_value = [
{
rule_type = NOT_NULL
}
]
}
]
}
}
}
Loading
Loading