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.

Sign up for GitHub

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

Jump to bottom

[Feature] support LZO compress on File Read #5083

Merged
merged 21 commits into from
Oct 16, 2023
Merged

[Feature] support LZO compress on File Read #5083

Show file tree
Hide file tree
Changes from 11 commits
Commits
Show all changes
21 commits
Select commit Hold shift + click to select a range
40f1993
[Feature] support compress on Text File Read
liunaijie Jul 12, 2023
e0e26b3
[Feature] fix code style
liunaijie Jul 14, 2023
21521a0
[Feature] update local file doc
liunaijie Jul 19, 2023
9ac95ca
[Improve] local file read compression
liunaijie Jul 20, 2023
5ba9c26
fix code style
liunaijie Jul 27, 2023
09c7e64
Merge branch 'dev' into text_file_read_compress
liunaijie Aug 1, 2023
f4a5d51
code style
liunaijie Aug 1, 2023
c2f4d6e
[Improve] add compress_codec parameter to file related connector
liunaijie Aug 9, 2023
86cb4c6
Merge branch 'up_dev' into text_file_read_compress
liunaijie Aug 12, 2023
36185a1
Merge branch 'dev' into text_file_read_compress
liunaijie Sep 4, 2023
3f966dc
code style fix
liunaijie Sep 4, 2023
01865f2
[Feature] add assert sink check
liunaijie Sep 15, 2023
1af0b31
[Feature] update assert sink check
liunaijie Sep 18, 2023
9c85559
Merge branch 'dev' into text_file_read_compress
liunaijie Sep 27, 2023
48552a6
Merge branch 'dev_' into pr/5083
Hisoka-X Oct 13, 2023
1445f96
update test case
Hisoka-X Oct 13, 2023
cadf6a4
Merge branch 'dev_' into pr/5083
Hisoka-X Oct 13, 2023
4a484ac
update
Hisoka-X Oct 14, 2023
ff83e7c
update
Hisoka-X Oct 14, 2023
6b6e388
Merge branch 'dev_' into pr/5083
Hisoka-X Oct 16, 2023
7e5cc2e
fix ci
Hisoka-X Oct 16, 2023
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
11 changes: 11 additions & 0 deletions docs/en/connector-v2/source/CosFile.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,6 +56,7 @@ Read all the data in a split in a pollNext call. What splits are read will be sa
| common-options | | no | - |
| sheet_name | string | no | - |
| file_filter_pattern | string | no | - |
| compress_codec | string | no | none |

### path [string]

Expand Down Expand Up @@ -252,6 +253,16 @@ Reader the sheet of the workbook,Only used when file_format is excel.

Filter pattern, which used for filtering files.

### compress_codec [string]

The compress codec of files and the details that supported as the following shown:

- txt: `lzo` `none`
- json: `lzo` `none`
- csv: `lzo` `none`
- orc/parquet:
automatically recognizes the compression type, no additional settings required.

## Example

```hocon
Expand Down
11 changes: 11 additions & 0 deletions docs/en/connector-v2/source/FtpFile.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,7 @@ If you use SeaTunnel Engine, It automatically integrated the hadoop jar when you
| common-options | | no | - |
| sheet_name | string | no | - |
| file_filter_pattern | string | no | - |
| compress_codec | string | no | none |

### host [string]

Expand Down Expand Up @@ -228,6 +229,16 @@ Source plugin common parameters, please refer to [Source Common Options](common-

Reader the sheet of the workbook,Only used when file_format_type is excel.

### compress_codec [string]

The compress codec of files and the details that supported as the following shown:

- txt: `lzo` `none`
- json: `lzo` `none`
- csv: `lzo` `none`
- orc/parquet:
automatically recognizes the compression type, no additional settings required.

## Example

```hocon
Expand Down
11 changes: 11 additions & 0 deletions docs/en/connector-v2/source/HdfsFile.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,6 +57,17 @@ Read data from hdfs file system.
| schema | config | no | - | the schema fields of upstream data |
| common-options | | no | - | Source plugin common parameters, please refer to [Source Common Options](common-options.md) for details. |
| sheet_name | string | no | - | Reader the sheet of the workbook,Only used when file_format is excel. |
| compress_codec | string | no | none | The compress codec of files |

### compress_codec [string]

The compress codec of files and the details that supported as the following shown:

- txt: `lzo` `none`
- json: `lzo` `none`
- csv: `lzo` `none`
- orc/parquet:
automatically recognizes the compression type, no additional settings required.

### Tips

Expand Down
11 changes: 11 additions & 0 deletions docs/en/connector-v2/source/Hive.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,7 @@ Read all the data in a split in a pollNext call. What splits are read will be sa
| read_partitions | list | no | - |
| read_columns | list | no | - |
| abort_drop_partition_metadata | boolean | no | true |
| compress_codec | string | no | none |
| common-options | | no | - |

### table_name [string]
Expand Down Expand Up @@ -85,6 +86,16 @@ The read column list of the data source, user can use it to implement field proj

Flag to decide whether to drop partition metadata from Hive Metastore during an abort operation. Note: this only affects the metadata in the metastore, the data in the partition will always be deleted(data generated during the synchronization process).

### compress_codec [string]

The compress codec of files and the details that supported as the following shown:

- txt: `lzo` `none`
- json: `lzo` `none`
- csv: `lzo` `none`
- orc/parquet:
automatically recognizes the compression type, no additional settings required.

### common options

Source plugin common parameters, please refer to [Source Common Options](common-options.md) for details
Expand Down
11 changes: 11 additions & 0 deletions docs/en/connector-v2/source/LocalFile.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,6 +50,7 @@ Read all the data in a split in a pollNext call. What splits are read will be sa
| common-options | | no | - |
| sheet_name | string | no | - |
| file_filter_pattern | string | no | - |
| compress_codec | string | no | none |

### path [string]

Expand Down Expand Up @@ -230,6 +231,16 @@ Reader the sheet of the workbook,Only used when file_format_type is excel.

Filter pattern, which used for filtering files.

### compress_codec [string]

The compress codec of files and the details that supported as the following shown:

- txt: `lzo` `none`
- json: `lzo` `none`
- csv: `lzo` `none`
- orc/parquet:
automatically recognizes the compression type, no additional settings required.

## Example

```hocon
Expand Down
11 changes: 11 additions & 0 deletions docs/en/connector-v2/source/OssFile.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,6 +57,7 @@ Read all the data in a split in a pollNext call. What splits are read will be sa
| common-options | | no | - |
| sheet_name | string | no | - |
| file_filter_pattern | string | no | - |
| compress_codec | string | no | none |

### path [string]

Expand Down Expand Up @@ -249,6 +250,16 @@ Source plugin common parameters, please refer to [Source Common Options](common-

Reader the sheet of the workbook,Only used when file_format_type is excel.

### compress_codec [string]

The compress codec of files and the details that supported as the following shown:

- txt: `lzo` `none`
- json: `lzo` `none`
- csv: `lzo` `none`
- orc/parquet:
automatically recognizes the compression type, no additional settings required.

## Example

```hocon
Expand Down
11 changes: 11 additions & 0 deletions docs/en/connector-v2/source/OssJindoFile.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,6 +57,7 @@ Read all the data in a split in a pollNext call. What splits are read will be sa
| common-options | | no | - |
| sheet_name | string | no | - |
| file_filter_pattern | string | no | - |
| compress_codec | string | no | none |

### path [string]

Expand Down Expand Up @@ -253,6 +254,16 @@ Reader the sheet of the workbook,Only used when file_format_type is excel.

Filter pattern, which used for filtering files.

### compress_codec [string]

The compress codec of files and the details that supported as the following shown:

- txt: `lzo` `none`
- json: `lzo` `none`
- csv: `lzo` `none`
- orc/parquet:
automatically recognizes the compression type, no additional settings required.

## Example

```hocon
Expand Down
11 changes: 11 additions & 0 deletions docs/en/connector-v2/source/S3File.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -214,6 +214,17 @@ If you assign file type to `parquet` `orc`, schema option not required, connecto
| schema | config | no | - | The schema of upstream data. |
| common-options | | no | - | Source plugin common parameters, please refer to [Source Common Options](common-options.md) for details. |
| sheet_name | string | no | - | Reader the sheet of the workbook,Only used when file_format is excel. |
| compress_codec | string | no | none |

### compress_codec [string]

The compress codec of files and the details that supported as the following shown:

- txt: `lzo` `none`
- json: `lzo` `none`
- csv: `lzo` `none`
- orc/parquet:
automatically recognizes the compression type, no additional settings required.

## Example

Expand Down
11 changes: 11 additions & 0 deletions docs/en/connector-v2/source/SftpFile.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,7 @@ If you use SeaTunnel Engine, It automatically integrated the hadoop jar when you
| common-options | | no | - |
| sheet_name | string | no | - |
| file_filter_pattern | string | no | - |
| compress_codec | string | no | none |

### host [string]

Expand Down Expand Up @@ -231,6 +232,16 @@ Reader the sheet of the workbook,Only used when file_format_type is excel.

Filter pattern, which used for filtering files.

### compress_codec [string]

The compress codec of files and the details that supported as the following shown:

- txt: `lzo` `none`
- json: `lzo` `none`
- csv: `lzo` `none`
- orc/parquet:
automatically recognizes the compression type, no additional settings required.

## Example

```hocon
Expand Down
6 changes: 6 additions & 0 deletions ...src/main/java/org/apache/seatunnel/connectors/seatunnel/file/config/BaseSourceConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -119,4 +119,10 @@ public class BaseSourceConfig {
.noDefaultValue()
.withDescription(
"File pattern. The connector will filter some files base on the pattern.");

public static final Option<CompressFormat> COMPRESS_CODEC =
Options.key("compress_codec")
.enumType(CompressFormat.class)
.defaultValue(CompressFormat.NONE)
.withDescription("Compression codec");
}
36 changes: 34 additions & 2 deletions ...n/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/JsonReadStrategy.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,8 @@
import org.apache.seatunnel.api.table.type.SeaTunnelRow;
import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
import org.apache.seatunnel.common.exception.CommonErrorCode;
import org.apache.seatunnel.connectors.seatunnel.file.config.BaseSourceConfig;
import org.apache.seatunnel.connectors.seatunnel.file.config.CompressFormat;
import org.apache.seatunnel.connectors.seatunnel.file.config.HadoopConf;
import org.apache.seatunnel.connectors.seatunnel.file.exception.FileConnectorException;
import org.apache.seatunnel.format.json.JsonDeserializationSchema;
Expand All @@ -30,14 +32,29 @@
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import io.airlift.compress.lzo.LzopCodec;
import lombok.extern.slf4j.Slf4j;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;
import java.util.Map;

@Slf4j
public class JsonReadStrategy extends AbstractReadStrategy {
private DeserializationSchema<SeaTunnelRow> deserializationSchema;
private CompressFormat compressFormat = BaseSourceConfig.COMPRESS_CODEC.defaultValue();

@Override
public void init(HadoopConf conf) {
super.init(conf);
if (pluginConfig.hasPath(BaseSourceConfig.COMPRESS_CODEC.key())) {
String compressCodec = pluginConfig.getString(BaseSourceConfig.COMPRESS_CODEC.key());
compressFormat = CompressFormat.valueOf(compressCodec.toUpperCase());
}
}

@Override
public void setSeaTunnelRowTypeInfo(SeaTunnelRowType seaTunnelRowType) {
Expand All @@ -58,9 +75,24 @@ public void read(String path, Collector<SeaTunnelRow> output)
FileSystem fs = FileSystem.get(conf);
Path filePath = new Path(path);
Map<String, String> partitionsMap = parsePartitionsByPath(path);
InputStream inputStream;
switch (compressFormat) {
case LZO:
LzopCodec lzo = new LzopCodec();
inputStream = lzo.createInputStream(fs.open(filePath));
break;
case NONE:
inputStream = fs.open(filePath);
break;
default:
log.warn(
"Text file does not support this compress type: {}",
compressFormat.getCompressCodec());
inputStream = fs.open(filePath);
break;
}
try (BufferedReader reader =
new BufferedReader(
new InputStreamReader(fs.open(filePath), StandardCharsets.UTF_8))) {
new BufferedReader(new InputStreamReader(inputStream, StandardCharsets.UTF_8))) {
reader.lines()
.forEach(
line -> {
Expand Down
31 changes: 29 additions & 2 deletions ...n/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/TextReadStrategy.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@
import org.apache.seatunnel.common.utils.DateUtils;
import org.apache.seatunnel.common.utils.TimeUtils;
import org.apache.seatunnel.connectors.seatunnel.file.config.BaseSourceConfig;
import org.apache.seatunnel.connectors.seatunnel.file.config.CompressFormat;
import org.apache.seatunnel.connectors.seatunnel.file.config.FileFormat;
import org.apache.seatunnel.connectors.seatunnel.file.config.HadoopConf;
import org.apache.seatunnel.connectors.seatunnel.file.exception.FileConnectorErrorCode;
Expand All @@ -39,19 +40,25 @@
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import io.airlift.compress.lzo.LzopCodec;
import lombok.extern.slf4j.Slf4j;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;
import java.util.Map;

@Slf4j
public class TextReadStrategy extends AbstractReadStrategy {
private DeserializationSchema<SeaTunnelRow> deserializationSchema;
private String fieldDelimiter = BaseSourceConfig.DELIMITER.defaultValue();
private DateUtils.Formatter dateFormat = BaseSourceConfig.DATE_FORMAT.defaultValue();
private DateTimeUtils.Formatter datetimeFormat =
BaseSourceConfig.DATETIME_FORMAT.defaultValue();
private TimeUtils.Formatter timeFormat = BaseSourceConfig.TIME_FORMAT.defaultValue();
private CompressFormat compressFormat = BaseSourceConfig.COMPRESS_CODEC.defaultValue();
private int[] indexes;

@Override
Expand All @@ -61,9 +68,25 @@ public void read(String path, Collector<SeaTunnelRow> output)
FileSystem fs = FileSystem.get(conf);
Path filePath = new Path(path);
Map<String, String> partitionsMap = parsePartitionsByPath(path);
InputStream inputStream;
switch (compressFormat) {
case LZO:
LzopCodec lzo = new LzopCodec();
inputStream = lzo.createInputStream(fs.open(filePath));
break;
case NONE:
inputStream = fs.open(filePath);
break;
default:
log.warn(
"Text file does not support this compress type: {}",
compressFormat.getCompressCodec());
inputStream = fs.open(filePath);
break;
}

try (BufferedReader reader =
new BufferedReader(
new InputStreamReader(fs.open(filePath), StandardCharsets.UTF_8))) {
new BufferedReader(new InputStreamReader(inputStream, StandardCharsets.UTF_8))) {
reader.lines()
.skip(skipHeaderNumber)
.forEach(
Expand Down Expand Up @@ -200,5 +223,9 @@ private void initFormatter() {
TimeUtils.Formatter.parse(
pluginConfig.getString(BaseSourceConfig.TIME_FORMAT.key()));
}
if (pluginConfig.hasPath(BaseSourceConfig.COMPRESS_CODEC.key())) {
String compressCodec = pluginConfig.getString(BaseSourceConfig.COMPRESS_CODEC.key());
compressFormat = CompressFormat.valueOf(compressCodec.toUpperCase());
}
}
}
1 change: 1 addition & 0 deletions .../java/org/apache/seatunnel/connectors/seatunnel/file/cos/source/CosFileSourceFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,7 @@ public OptionRule optionRule() {
.optional(BaseSourceConfig.DATETIME_FORMAT)
.optional(BaseSourceConfig.TIME_FORMAT)
.optional(BaseSourceConfig.FILE_FILTER_PATTERN)
.optional(BaseSourceConfig.COMPRESS_CODEC)
.build();
}

Expand Down
1 change: 1 addition & 0 deletions .../java/org/apache/seatunnel/connectors/seatunnel/file/ftp/source/FtpFileSourceFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,7 @@ public OptionRule optionRule() {
.optional(BaseSourceConfig.DATETIME_FORMAT)
.optional(BaseSourceConfig.TIME_FORMAT)
.optional(BaseSourceConfig.FILE_FILTER_PATTERN)
.optional(BaseSourceConfig.COMPRESS_CODEC)
.build();
}

Expand Down
1 change: 1 addition & 0 deletions ...ava/org/apache/seatunnel/connectors/seatunnel/file/hdfs/source/HdfsFileSourceFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,7 @@ public OptionRule optionRule() {
.optional(BaseSourceConfig.DATETIME_FORMAT)
.optional(BaseSourceConfig.TIME_FORMAT)
.optional(BaseSourceConfig.FILE_FILTER_PATTERN)
.optional(BaseSourceConfig.COMPRESS_CODEC)
.build();
}

Expand Down
1 change: 1 addition & 0 deletions .../java/org/apache/seatunnel/connectors/seatunnel/file/oss/source/OssFileSourceFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,7 @@ public OptionRule optionRule() {
.optional(BaseSourceConfig.DATETIME_FORMAT)
.optional(BaseSourceConfig.TIME_FORMAT)
.optional(BaseSourceConfig.FILE_FILTER_PATTERN)
.optional(BaseSourceConfig.COMPRESS_CODEC)
.build();
}

Expand Down
1 change: 1 addition & 0 deletions .../java/org/apache/seatunnel/connectors/seatunnel/file/oss/source/OssFileSourceFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,7 @@ public OptionRule optionRule() {
.optional(BaseSourceConfig.DATETIME_FORMAT)
.optional(BaseSourceConfig.TIME_FORMAT)
.optional(BaseSourceConfig.FILE_FILTER_PATTERN)
.optional(BaseSourceConfig.COMPRESS_CODEC)
.build();
}

Expand Down
Loading
Loading