import ChangeLog from '../changelog/connector-file-hadoop.md';

HdfsFile

Hdfs File Source Connector

Support Those Engines

Spark

Flink

SeaTunnel Zeta

Key Features

• batch

• stream

• multimodal

  Use binary file format to read and write files in any format, such as videos, pictures, etc. In short, any files can be synchronized to the target place.

• exactly-once

  Read all the data in a split in a pollNext call. What splits are read will be saved in snapshot.

• column projection

• parallelism

• support user-defined split

• support multiple table read

• file format file

  • text
  • csv
  • parquet
  • orc
  • json
  • excel
  • xml
  • binary
  • markdown

Description

Read data from hdfs file system.

Supported DataSource Info

Source Options

file_format_type [string]

File type, supported as the following file types:

text csv parquet orc json excel xml binary markdown

If you assign file type to markdown, SeaTunnel can parse markdown files and extract structured data. The markdown parser extracts various elements including headings, paragraphs, lists, code blocks, tables, and more. Each element is converted to a row with the following schema:

• element_id: Unique identifier for the element
• element_type: Type of the element (Heading, Paragraph, ListItem, etc.)
• heading_level: Level of heading (1-6, null for non-heading elements)
• text: Text content of the element
• page_number: Page number (default: 1)
• position_index: Position index within the document
• parent_id: ID of the parent element
• child_ids: Comma-separated list of child element IDs

Note: Markdown format only supports reading, not writing.

delimiter/field_delimiter [string]

delimiter parameter will deprecate after version 2.3.5, please use field_delimiter instead.

row_delimiter [string]

Only need to be configured when file_format is text

Row delimiter, used to tell connector how to slice and dice rows

default \n

file_filter_pattern [string]

Filter pattern, which used for filtering files. If you only want to filter based on file names, simply write the regular file names; If you want to filter based on the file directory at the same time, the expression needs to start with path.

The pattern follows standard regular expressions. For details, please refer to https://en.wikipedia.org/wiki/Regular_expression. There are some examples.

If the path is /data/seatunnel, and the file structure example is:

/data/seatunnel/20241001/report.txt
/data/seatunnel/20241007/abch202410.csv
/data/seatunnel/20241002/abcg202410.csv
/data/seatunnel/20241005/old_data.csv
/data/seatunnel/20241012/logo.png

Matching Rules Example:

Example 1: Match all .txt files，Regular Expression:

.*.txt

The result of this example matching is:

/data/seatunnel/20241001/report.txt

Example 2: Match all file starting with abc，Regular Expression:

abc.*

The result of this example matching is:

/data/seatunnel/20241007/abch202410.csv
/data/seatunnel/20241002/abcg202410.csv

Example 3: Match all files starting with abc in folder 20241007，And the fourth character is either h or g, the Regular Expression:

/data/seatunnel/20241007/abc[h,g].*

The result of this example matching is:

/data/seatunnel/20241007/abch202410.csv

Example 4: Match third level folders starting with 202410 and files ending with .csv, the Regular Expression:

/data/seatunnel/202410\d*/.*.csv

The result of this example matching is:

/data/seatunnel/20241007/abch202410.csv
/data/seatunnel/20241002/abcg202410.csv
/data/seatunnel/20241005/old_data.csv

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.

archive_compress_codec [string]

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

Note: gz compressed excel file needs to compress the original file or specify the file suffix, such as e2e.xls ->e2e_test.xls.gz

encoding [string]

Only used when file_format_type is json,text,csv,xml. The encoding of the file to read. This param will be parsed by Charset.forName(encoding).

binary_chunk_size [int]

Only used when file_format_type is binary.

The chunk size (in bytes) for reading binary files. Default is 1024 bytes. Larger values may improve performance for large files but use more memory.

binary_complete_file_mode [boolean]

Only used when file_format_type is binary.

Whether to read the complete file as a single chunk instead of splitting into chunks. When enabled, the entire file content will be read into memory at once. Default is false.

sync_mode [string]

File sync mode. Supported values: full (default), update.

When sync_mode=update, the source will compare files between source/target and only read new/changed files (currently only supports file_format_type=binary).

target_path [string]

Only used when sync_mode=update.

Target base path used for comparison (it should usually be the same as sink path).

target_hadoop_conf [map]

Only used when sync_mode=update.

Extra Hadoop configuration for target filesystem (optional). If not set, it reuses the source filesystem configuration.

You can set fs.defaultFS in this map to override target defaultFS, e.g. "fs.defaultFS" = "hdfs://nn2:9000".

update_strategy [string]

Only used when sync_mode=update. Supported values: distcp (default), strict.

• distcp: similar to distcp -update:
  • target file not exists → COPY
  • length differs → COPY
  • mtime(source) > mtime(target) → COPY
  • else → SKIP
• strict: strict consistency, decided by compare_mode.

compare_mode [string]

Only used when sync_mode=update. Supported values: len_mtime (default), checksum.

• len_mtime: SKIP only when both len and mtime are equal, otherwise COPY.
• checksum: SKIP only when len is equal and Hadoop getFileChecksum is equal, otherwise COPY (only valid when update_strategy=strict).

enable_file_split [boolean]

Turn on the file splitting function, the default is false. It can be selected when the file type is csv, text, json, parquet and non-compressed format.

• text/csv/json: split by file_split_size and align to the next row_delimiter to avoid breaking records.
• parquet: split by RowGroup (logical split), never breaks a RowGroup.

Recommendations

• Enable when reading a few large files and you want higher read parallelism.
• Disable when reading many small files, or when parallelism is low (splitting adds overhead).

Limitations

• Not supported for compressed files (compress_codec != none) or archive files (archive_compress_codec != none) — it will fall back to non-splitting.
• For text/csv/json, actual split size may be larger than file_split_size because the split end is aligned to the next row_delimiter.

file_split_size [long]

File split size, which can be filled in when the enable_file_split parameter is true. The unit is the number of bytes. The default value is the number of bytes of 128MB, which is 134217728.

Tuning

• Start with the default (128MB). Decrease it if parallelism is under-utilized; increase it if the number of splits is too large.
• Rough rule: file_split_size ≈ file_size / desired_parallelism.

quote_char [string]

A single character that encloses CSV fields, allowing fields with commas, line breaks, or quotes to be read correctly.

escape_char [string]

A single character that allows the quote or other special characters to appear inside a CSV field without ending the field.

Tips

If you use spark/flink, In order to use this connector, You must ensure your spark/flink cluster already integrated hadoop. The tested hadoop version is 2.x. If you use SeaTunnel Engine, It automatically integrated the hadoop jar when you download and install SeaTunnel Engine. You can check the jar package under ${SEATUNNEL_HOME}/lib to confirm this.

Task Example

Simple

This example defines a SeaTunnel synchronization task that read data from Hdfs and sends it to Hdfs.

# Defining the runtime environment
env {
  parallelism = 1
  job.mode = "BATCH"
}

source {
  HdfsFile {
  schema {
    fields {
      name = string
      age = int
    }
  }
  path = "/apps/hive/demo/student"
  file_format_type = "json"
  fs.defaultFS = "hdfs://namenode001"
  }
  # If you would like to get more information about how to configure seatunnel and see full list of source plugins,
  # please go to https://seatunnel.apache.org/docs/connectors/source
}

transform {
  # 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/transforms
}

sink {
    HdfsFile {
      fs.defaultFS = "hdfs://hadoopcluster"
      path = "/tmp/hive/warehouse/test2"
      file_format_type = "orc"
    }
  # If you would like to get more information about how to configure seatunnel and see full list of sink plugins,
  # please go to https://seatunnel.apache.org/docs/connectors/sink
}

Filter File

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

source {
  HdfsFile {
    path = "/apps/hive/demo/student"
    file_format_type = "json"
    fs.defaultFS = "hdfs://namenode001"
    // file example abcD2024.csv
    file_filter_pattern = "abc[DX]*.*"
  }
}

sink {
  Console {
  }
}

Multiple Table

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

source {
  HdfsFile {
    tables_configs = [
      {
        schema = {
          table = "student"
        }
        path = "/apps/hive/demo/student"
        file_format_type = "json"
        fs.defaultFS = "hdfs://namenode001"
      },
      {
        schema = {
          table = "teacher"
        }
        path = "/apps/hive/demo/teacher"
        file_format_type = "json"
        fs.defaultFS = "hdfs://namenode001"
      }
    ]
  }
}

sink {
    HdfsFile {
      fs.defaultFS = "hdfs://hadoopcluster"
      path = "/tmp/hive/warehouse/${table_name}"
      file_format_type = "orc"
    }
}

Changelog