This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new b6cdd24087 [Feature][File] Add pdf parser for RAG support (#10105)
b6cdd24087 is described below
commit b6cdd24087c7417502cc5562a229555e37a00d78
Author: JeremyXin <[email protected]>
AuthorDate: Sun Jul 19 09:48:15 2026 +0800
[Feature][File] Add pdf parser for RAG support (#10105)
Co-authored-by: JeremyXin <[email protected]>
Co-authored-by: davidzollo <[email protected]>
---
.github/workflows/backend.yml | 2 +-
docs/en/connectors/source/CosFile.md | 16 +-
docs/en/connectors/source/FtpFile.md | 16 +-
docs/en/connectors/source/HdfsFile.md | 18 +-
docs/en/connectors/source/Hive.md | 2 +-
docs/en/connectors/source/LocalFile.md | 41 +-
docs/en/connectors/source/ObsFile.md | 16 +-
docs/en/connectors/source/OssFile.md | 18 +-
docs/en/connectors/source/OssJindoFile.md | 16 +-
docs/en/connectors/source/S3File.md | 18 +-
docs/en/connectors/source/SftpFile.md | 18 +-
docs/zh/connectors/source/CosFile.md | 17 +-
docs/zh/connectors/source/FtpFile.md | 17 +-
docs/zh/connectors/source/HdfsFile.md | 18 +-
docs/zh/connectors/source/Hive.md | 2 +-
docs/zh/connectors/source/LocalFile.md | 41 +-
docs/zh/connectors/source/ObsFile.md | 14 +-
docs/zh/connectors/source/OssFile.md | 18 +-
docs/zh/connectors/source/OssJindoFile.md | 15 +-
docs/zh/connectors/source/S3File.md | 18 +-
docs/zh/connectors/source/SftpFile.md | 17 +-
.../api/file/AllFileSpecificationCheckTest.java | 3 +-
.../connector-file/connector-file-base/pom.xml | 7 +
.../seatunnel/file/config/DocumentElement.java | 193 ++++
.../seatunnel/file/config/FileFormat.java | 13 +-
.../file/source/reader/PdfReadStrategy.java | 1034 ++++++++++++++++++++
.../file/source/reader/PdfReadStrategyTest.java | 348 +++++++
.../src/test/resources/pdf_read_strategy_test.pdf | Bin 0 -> 25289 bytes
28 files changed, 1910 insertions(+), 46 deletions(-)
diff --git a/.github/workflows/backend.yml b/.github/workflows/backend.yml
index 6d2202248d..6ec74cd558 100644
--- a/.github/workflows/backend.yml
+++ b/.github/workflows/backend.yml
@@ -447,7 +447,7 @@ jobs:
run: |
./mvnw -B -T 1 clean verify -DskipUT=false -DskipIT=true
-D"license.skipAddThirdParty"=true -D"skip.ui"=true --no-snapshot-updates
env:
- MAVEN_OPTS: -Xmx4096m
+ MAVEN_OPTS: -Xmx4096m -Dfile.encoding=UTF-8
updated-modules-integration-test-part-1:
needs: [ changes, sanity-check ]
diff --git a/docs/en/connectors/source/CosFile.md
b/docs/en/connectors/source/CosFile.md
index 560d8f4a38..380b5fc9f2 100644
--- a/docs/en/connectors/source/CosFile.md
+++ b/docs/en/connectors/source/CosFile.md
@@ -35,6 +35,7 @@ import ChangeLog from '../changelog/connector-file-cos.md';
- [x] xml
- [x] binary
- [x] markdown
+ - [x] pdf
## Description
@@ -96,7 +97,7 @@ The source file path.
File type, supported as the following file types:
-`text` `csv` `parquet` `orc` `json` `excel` `xml` `binary` `markdown`
+`text` `csv` `parquet` `orc` `json` `excel` `xml` `binary` `markdown` `pdf`
If you assign file type to `json`, you should also assign schema option to
tell connector how to parse data to the row you want.
@@ -187,7 +188,7 @@ at the same time. You can find the specific usage in the
example below.
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:
+Each extracted element is converted to a document-element 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)
@@ -208,6 +209,17 @@ The option defaults to `false`, so the original Markdown
schema is unchanged unl
Note: Markdown format only supports reading, not writing.
+If you assign file type to `pdf`, SeaTunnel can parse PDF files and extract
structured document elements.
+PDF uses the same document-element row schema described above.
+
+The main PDF-specific behaviors are:
+
+- **With outline**: Extracts `heading`, `paragraph`, `image`, and `link`
elements. Headings are derived from the outline structure, and elements are
organized into a parent-child hierarchy reflecting the document's logical
structure.
+- **Without outline**: Extracts only `paragraph` and `image` elements in a
flat structure without hierarchy.
+- `element_type` values for PDF are `heading`, `paragraph`, `image`, and
`link`.
+
+Note: Only single-column (top-to-bottom) PDF layouts are supported.
Multi-column layouts (e.g., side-by-side two-column documents) are not
supported and may produce incorrect text ordering.
+
### bucket [string]
The bucket address of COS file system, for example: `cosn://seatunnel-test`
diff --git a/docs/en/connectors/source/FtpFile.md
b/docs/en/connectors/source/FtpFile.md
index 18e22d344b..9da83a14ea 100644
--- a/docs/en/connectors/source/FtpFile.md
+++ b/docs/en/connectors/source/FtpFile.md
@@ -31,6 +31,7 @@ import ChangeLog from '../changelog/connector-file-ftp.md';
- [x] xml
- [x] binary
- [x] markdown
+ - [x] pdf
## Description
@@ -203,7 +204,7 @@ Filter filename extension, which used for filtering files
with specific extensio
File type, supported as the following file types:
-`text` `csv` `parquet` `orc` `json` `excel` `xml` `binary`
+`text` `csv` `parquet` `orc` `json` `excel` `xml` `binary` `markdown` `pdf`
If you assign file type to `json` , you should also assign schema option to
tell connector how to parse data to the row you want.
@@ -283,7 +284,7 @@ at the same time. You can find the specific usage in the
example below.
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:
+Each extracted element is converted to a document-element 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)
@@ -304,6 +305,17 @@ The option defaults to `false`, so the original Markdown
schema is unchanged unl
Note: Markdown format only supports reading, not writing.
+If you assign file type to `pdf`, SeaTunnel can parse PDF files and extract
structured document elements.
+PDF uses the same document-element row schema described above.
+
+The main PDF-specific behaviors are:
+
+- **With outline**: Extracts `heading`, `paragraph`, `image`, and `link`
elements. Headings are derived from the outline structure, and elements are
organized into a parent-child hierarchy reflecting the document's logical
structure.
+- **Without outline**: Extracts only `paragraph` and `image` elements in a
flat structure without hierarchy.
+- `element_type` values for PDF are `heading`, `paragraph`, `image`, and
`link`.
+
+Note: Only single-column (top-to-bottom) PDF layouts are supported.
Multi-column layouts (e.g., side-by-side two-column documents) are not
supported and may produce incorrect text ordering.
+
### connection_mode [string]
The target ftp connection mode , default is active mode, supported as the
following modes:
diff --git a/docs/en/connectors/source/HdfsFile.md
b/docs/en/connectors/source/HdfsFile.md
index c71079a696..fd28d44d68 100644
--- a/docs/en/connectors/source/HdfsFile.md
+++ b/docs/en/connectors/source/HdfsFile.md
@@ -36,6 +36,7 @@ import ChangeLog from '../changelog/connector-file-hadoop.md';
- [x] xml
- [x] binary
- [x] markdown
+ - [x] pdf
## Description
@@ -53,7 +54,7 @@ Read data from hdfs file system.
|----------------------------|---------|----------|-----------------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| path | string | yes | -
| The source file path.
|
| tables_configs | list | no | -
| Configure multiple HDFS source tables in one source block. Each item has the
same options as a single-table `HdfsFile` source, and can set its downstream
table name through `schema.table`.
|
-| file_format_type | string | yes | -
| We supported as the following file types:`text` `csv` `parquet` `orc` `json`
`excel` `xml` `binary` `markdown`.Please note that, The final file name will
end with the file_format's suffix, the suffix of the text file is `txt`.
|
+| file_format_type | string | yes | -
| We supported as the following file types:`text` `csv` `parquet` `orc` `json`
`excel` `xml` `binary` `markdown` `pdf`.Please note that, The final file name
will end with the file_format's suffix, the suffix of the text file is `txt`.
|
| fs.defaultFS | string | yes | -
| The hadoop cluster address that start with `hdfs://`, for example:
`hdfs://hadoopcluster`
|
| read_columns | list | no | -
| The read column list of the data source, user can use it to implement field
projection.The file type supported column projection as the following
shown:[text,json,csv,orc,parquet,excel,xml].Tips: If the user wants to use this
feature when reading `text` `json` `csv` files, the schema option must be
configured. |
| hdfs_site_path | string | no | -
| The path of `hdfs-site.xml`, used to load ha configuration of namenodes
|
@@ -104,11 +105,11 @@ Read data from hdfs file system.
File type, supported as the following file types:
-`text` `csv` `parquet` `orc` `json` `excel` `xml` `binary` `markdown`
+`text` `csv` `parquet` `orc` `json` `excel` `xml` `binary` `markdown` `pdf`
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:
+Each extracted element is converted to a document-element 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)
@@ -129,6 +130,17 @@ The option defaults to `false`, so the original Markdown
schema is unchanged unl
Note: Markdown format only supports reading, not writing.
+If you assign file type to `pdf`, SeaTunnel can parse PDF files and extract
structured document elements.
+PDF uses the same document-element row schema described above.
+
+The main PDF-specific behaviors are:
+
+- **With outline**: Extracts `heading`, `paragraph`, `image`, and `link`
elements. Headings are derived from the outline structure, and elements are
organized into a parent-child hierarchy reflecting the document's logical
structure.
+- **Without outline**: Extracts only `paragraph` and `image` elements in a
flat structure without hierarchy.
+- `element_type` values for PDF are `heading`, `paragraph`, `image`, and
`link`.
+
+Note: Only single-column (top-to-bottom) PDF layouts are supported.
Multi-column layouts (e.g., side-by-side two-column documents) are not
supported and may produce incorrect text ordering.
+
### tables_configs [list]
Use `tables_configs` when one HDFS source needs to read multiple tables or
directories. Each item can define its own
diff --git a/docs/en/connectors/source/Hive.md
b/docs/en/connectors/source/Hive.md
index b5504bebac..50c07f3273 100644
--- a/docs/en/connectors/source/Hive.md
+++ b/docs/en/connectors/source/Hive.md
@@ -8,7 +8,7 @@ import ChangeLog from '../changelog/connector-hive.md';
Read data from Hive.
-When using markdown format, SeaTunnel can parse markdown files stored in Hive
tables and extract structured data with elements like headings, paragraphs,
lists, code blocks, and tables. Each element is converted to a row with the
following schema:
+When using markdown format, SeaTunnel can parse markdown files stored in Hive
tables and extract structured data with elements like headings, paragraphs,
lists, code blocks, and tables. Each extracted element is converted to a
document-element 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)
diff --git a/docs/en/connectors/source/LocalFile.md
b/docs/en/connectors/source/LocalFile.md
index 7e9985f533..abda9935fd 100644
--- a/docs/en/connectors/source/LocalFile.md
+++ b/docs/en/connectors/source/LocalFile.md
@@ -35,6 +35,7 @@ import ChangeLog from '../changelog/connector-file-local.md';
- [x] xml
- [x] binary
- [x] markdown
+ - [x] pdf
## Description
@@ -104,7 +105,7 @@ The source file path.
File type, supported as the following file types:
-`text` `csv` `parquet` `orc` `json` `excel` `xml` `binary` `markdown`
+`text` `csv` `parquet` `orc` `json` `excel` `xml` `binary` `markdown` `pdf`
If you assign file type to `json`, you should also assign schema option to
tell connector how to parse data to the row you want.
@@ -195,7 +196,7 @@ at the same time. You can find the specific usage in the
example below.
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:
+Each extracted element is converted to a document-element 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)
@@ -216,6 +217,17 @@ The option defaults to `false`, so the original Markdown
schema is unchanged unl
Note: Markdown format only supports reading, not writing.
+If you assign file type to `pdf`, SeaTunnel can parse PDF files and extract
structured document elements.
+PDF uses the same document-element row schema described above.
+
+The main PDF-specific behaviors are:
+
+- **With outline**: Extracts `heading`, `paragraph`, `image`, and `link`
elements. Headings are derived from the outline structure, and elements are
organized into a parent-child hierarchy reflecting the document's logical
structure.
+- **Without outline**: Extracts only `paragraph` and `image` elements in a
flat structure without hierarchy.
+- `element_type` values for PDF are `heading`, `paragraph`, `image`, and
`link`.
+
+Note: Only single-column (top-to-bottom) PDF layouts are supported.
Multi-column layouts (e.g., side-by-side two-column documents) are not
supported and may produce incorrect text ordering.
+
### read_columns [list]
The read column list of the data source, user can use it to implement field
projection.
@@ -652,6 +664,31 @@ LocalFile {
```
+### Read PDF File
+
+```hocon
+
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ LocalFile {
+ path = "/data/documents/"
+ file_format_type = "pdf"
+ }
+}
+
+sink {
+ Console {
+ }
+}
+
+```
+
+For best results, use PDF files that contain an outline (bookmarks/table of
contents). This enables the parser to extract headings with hierarchy
information.
+
### Transfer Binary File
```hocon
diff --git a/docs/en/connectors/source/ObsFile.md
b/docs/en/connectors/source/ObsFile.md
index 0f833701d1..70df651de0 100644
--- a/docs/en/connectors/source/ObsFile.md
+++ b/docs/en/connectors/source/ObsFile.md
@@ -35,6 +35,7 @@ import ChangeLog from '../changelog/connector-file-obs.md';
- [x] json
- [x] excel
- [x] markdown
+ - [x] pdf
## Description
@@ -143,7 +144,7 @@ It only supports hadoop version **2.9.X+**.
> File type, supported as the following file types:
>
-> `text` `csv` `parquet` `orc` `json` `excel` `markdown`
+> `text` `csv` `parquet` `orc` `json` `excel` `xml` `binary` `markdown` `pdf`
>
> If you assign file type to `json`, you should also assign schema option to
> tell the connector how to parse data to the row you want.
>
@@ -225,7 +226,7 @@ schema {
> 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:
+> Each extracted element is converted to a document-element 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)
@@ -245,6 +246,17 @@ schema {
> The option defaults to `false`, so the original Markdown schema is unchanged
> unless you enable it.
>
> Note: Markdown format only supports reading, not writing.
+>
+> If you assign file type to `pdf`, SeaTunnel can parse PDF files and extract
structured document elements.
+> PDF uses the same document-element row schema described above.
+>
+> The main PDF-specific behaviors are:
+>
+> - **With outline**: Extracts `heading`, `paragraph`, `image`, and `link`
elements. Headings are derived from the outline structure, and elements are
organized into a parent-child hierarchy reflecting the document's logical
structure.
+> - **Without outline**: Extracts only `paragraph` and `image` elements in a
flat structure without hierarchy.
+> - `element_type` values for PDF are `heading`, `paragraph`, `image`, and
`link`.
+>
+> Note: Only single-column (top-to-bottom) PDF layouts are supported.
Multi-column layouts (e.g., side-by-side two-column documents) are not
supported and may produce incorrect text ordering.
#### <span id="schema"> schema </span>
diff --git a/docs/en/connectors/source/OssFile.md
b/docs/en/connectors/source/OssFile.md
index 077907ab42..c85f9859b4 100644
--- a/docs/en/connectors/source/OssFile.md
+++ b/docs/en/connectors/source/OssFile.md
@@ -46,6 +46,7 @@ import ChangeLog from '../changelog/connector-file-oss.md';
- [x] xml
- [x] binary
- [x] markdown
+ - [x] pdf
## Data Type Mapping
@@ -186,7 +187,7 @@ If you assign file type to `parquet` `orc`, schema option
not required, connecto
| name | type | required | default value |
Description
|
|----------------------------|---------|----------|---------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| path | string | yes | - | The
Oss path that needs to be read can have sub paths, but the sub paths need to
meet certain format requirements. Specific requirements can be referred to
"parse_partition_from_path" option
|
-| file_format_type | string | yes | - | File
type, supported as the following file types: `text` `csv` `parquet` `orc`
`json` `excel` `xml` `binary` `markdown`
|
+| file_format_type | string | yes | - | File
type, supported as the following file types: `text` `csv` `parquet` `orc`
`json` `excel` `xml` `binary` `markdown` `pdf`
|
| bucket | string | yes | - | The
bucket address of oss file system, for example: `oss://seatunnel-test`.
|
| endpoint | string | yes | - | fs
oss endpoint
|
| read_columns | list | no | - | The
read column list of the data source, user can use it to implement field
projection. The file type supported column projection as the following shown:
`text` `csv` `parquet` `orc` `json` `excel` `xml` . If the user wants to use
this feature when reading `text` `json` `csv` files, the "schema" option must
be configured. |
@@ -224,11 +225,11 @@ If you assign file type to `parquet` `orc`, schema option
not required, connecto
File type, supported as the following file types:
-`text` `csv` `parquet` `orc` `json` `excel` `xml` `binary` `markdown`
+`text` `csv` `parquet` `orc` `json` `excel` `xml` `binary` `markdown` `pdf`
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:
+Each extracted element is converted to a document-element 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)
@@ -249,6 +250,17 @@ The option defaults to `false`, so the original Markdown
schema is unchanged unl
Note: Markdown format only supports reading, not writing.
+If you assign file type to `pdf`, SeaTunnel can parse PDF files and extract
structured document elements.
+PDF uses the same document-element row schema described above.
+
+The main PDF-specific behaviors are:
+
+- **With outline**: Extracts `heading`, `paragraph`, `image`, and `link`
elements. Headings are derived from the outline structure, and elements are
organized into a parent-child hierarchy reflecting the document's logical
structure.
+- **Without outline**: Extracts only `paragraph` and `image` elements in a
flat structure without hierarchy.
+- `element_type` values for PDF are `heading`, `paragraph`, `image`, and
`link`.
+
+Note: Only single-column (top-to-bottom) PDF layouts are supported.
Multi-column layouts (e.g., side-by-side two-column documents) are not
supported and may produce incorrect text ordering.
+
### compress_codec [string]
The compress codec of files and the details that supported as the following
shown:
diff --git a/docs/en/connectors/source/OssJindoFile.md
b/docs/en/connectors/source/OssJindoFile.md
index 7a4f965480..6764d7a523 100644
--- a/docs/en/connectors/source/OssJindoFile.md
+++ b/docs/en/connectors/source/OssJindoFile.md
@@ -35,6 +35,7 @@ import ChangeLog from
'../changelog/connector-file-oss-jindo.md';
- [x] xml
- [x] binary
- [x] markdown
+ - [x] pdf
## Description
@@ -98,7 +99,7 @@ The source file path.
File type, supported as the following file types:
-`text` `csv` `parquet` `orc` `json` `excel` `xml` `binary` `markdown`
+`text` `csv` `parquet` `orc` `json` `excel` `xml` `binary` `markdown` `pdf`
If you assign file type to `json`, you should also assign schema option to
tell connector how to parse data to the row you want.
@@ -189,7 +190,7 @@ at the same time. You can find the specific usage in the
example below.
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:
+Each extracted element is converted to a document-element 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)
@@ -210,6 +211,17 @@ The option defaults to `false`, so the original Markdown
schema is unchanged unl
Note: Markdown format only supports reading, not writing.
+If you assign file type to `pdf`, SeaTunnel can parse PDF files and extract
structured document elements.
+PDF uses the same document-element row schema described above.
+
+The main PDF-specific behaviors are:
+
+- **With outline**: Extracts `heading`, `paragraph`, `image`, and `link`
elements. Headings are derived from the outline structure, and elements are
organized into a parent-child hierarchy reflecting the document's logical
structure.
+- **Without outline**: Extracts only `paragraph` and `image` elements in a
flat structure without hierarchy.
+- `element_type` values for PDF are `heading`, `paragraph`, `image`, and
`link`.
+
+Note: Only single-column (top-to-bottom) PDF layouts are supported.
Multi-column layouts (e.g., side-by-side two-column documents) are not
supported and may produce incorrect text ordering.
+
### bucket [string]
The bucket address of oss file system, for example:
`oss://tyrantlucifer-image-bed`
diff --git a/docs/en/connectors/source/S3File.md
b/docs/en/connectors/source/S3File.md
index 71bd4cedd9..6b510045e1 100644
--- a/docs/en/connectors/source/S3File.md
+++ b/docs/en/connectors/source/S3File.md
@@ -35,6 +35,7 @@ import ChangeLog from '../changelog/connector-file-s3.md';
- [x] xml
- [x] binary
- [x] markdown
+ - [x] pdf
## Description
@@ -192,7 +193,7 @@ If you assign file type to `parquet` `orc`, schema option
not required, connecto
| name | type | required | default value
| Description
[...]
|---------------------------------|---------|----------|-------------------------------------------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
[...]
| path | string | yes | -
| The s3 path that needs to be read can have
sub paths, but the sub paths need to meet certain format requirements. Specific
requirements can be referred to "parse_partition_from_path" option
[...]
-| file_format_type | string | yes | -
| File type, supported as the following file
types: `text` `csv` `parquet` `orc` `json` `excel` `xml` `binary` `markdown`
[...]
+| file_format_type | string | yes | -
| File type, supported as the following file
types: `text` `csv` `parquet` `orc` `json` `excel` `xml` `binary` `markdown`
`pdf`
[...]
| bucket | string | yes | -
| The bucket address of s3 file system, for
example: `s3n://seatunnel-test`, if you use `s3a` protocol, this parameter
should be `s3a://seatunnel-test`.
[...]
| fs.s3a.endpoint | string | yes | -
| fs s3a endpoint
[...]
| fs.s3a.aws.credentials.provider | string | yes |
com.amazonaws.auth.InstanceProfileCredentialsProvider | The way to authenticate
s3a. We only support `org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider`
and `com.amazonaws.auth.InstanceProfileCredentialsProvider` now. More
information about the credential provider you can see [Hadoop AWS
Document](https://hadoop.apache.org/docs/stable/hadoop-aws/tools/hadoop-aws/index.html#Simple_name.2Fsecret_credentials_with_SimpleAWSCredenti
[...]
@@ -234,11 +235,11 @@ If you assign file type to `parquet` `orc`, schema option
not required, connecto
File type, supported as the following file types:
-`text` `csv` `parquet` `orc` `json` `excel` `xml` `binary` `markdown`
+`text` `csv` `parquet` `orc` `json` `excel` `xml` `binary` `markdown` `pdf`
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:
+Each extracted element is converted to a document-element 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)
@@ -259,6 +260,17 @@ The option defaults to `false`, so the original Markdown
schema is unchanged unl
Note: Markdown format only supports reading, not writing.
+If you assign file type to `pdf`, SeaTunnel can parse PDF files and extract
structured document elements.
+PDF uses the same document-element row schema described above.
+
+The main PDF-specific behaviors are:
+
+- **With outline**: Extracts `heading`, `paragraph`, `image`, and `link`
elements. Headings are derived from the outline structure, and elements are
organized into a parent-child hierarchy reflecting the document's logical
structure.
+- **Without outline**: Extracts only `paragraph` and `image` elements in a
flat structure without hierarchy.
+- `element_type` values for PDF are `heading`, `paragraph`, `image`, and
`link`.
+
+Note: Only single-column (top-to-bottom) PDF layouts are supported.
Multi-column layouts (e.g., side-by-side two-column documents) are not
supported and may produce incorrect text ordering.
+
### delimiter/field_delimiter [string]
**delimiter** parameter will deprecate after version 2.3.5, please use
**field_delimiter** instead.
diff --git a/docs/en/connectors/source/SftpFile.md
b/docs/en/connectors/source/SftpFile.md
index 92bee70989..1c64f11ec7 100644
--- a/docs/en/connectors/source/SftpFile.md
+++ b/docs/en/connectors/source/SftpFile.md
@@ -30,6 +30,7 @@ import ChangeLog from '../changelog/connector-file-sftp.md';
- [x] xml
- [x] binary
- [x] markdown
+ - [x] pdf
## Description
@@ -86,7 +87,7 @@ The File does not have a specific type list, and we can
indicate which SeaTunnel
| password | String | No | -
| The target sftp password. Required when `keyfile` is not set.
|
| keyfile | String | No | -
| The private key file path used for SFTP public key authentication.
|
| path | String | Yes | -
| The source file path.
|
-| file_format_type | String | Yes | -
| Please check #file_format_type below
|
+| file_format_type | String | Yes | -
| Please check #file_format_type below. Supported file types: `text` `csv`
`parquet` `orc` `json` `excel` `xml` `binary` `markdown` `pdf`
[...]
| file_filter_pattern | String | No | -
| Filter pattern, which used for filtering files.
|
| filename_extension | string | no | -
| Filter filename extension, which used for filtering files with specific
extension. Example: `csv` `.txt` `json` `.xml`.
|
| delimiter/field_delimiter | String | No | \001 for text and ',' for
csv | **delimiter** parameter will deprecate after version 2.3.5, please use
**field_delimiter** instead. <br/> Field delimiter, used to tell connector how
to slice and dice fields when reading text files. <br/> Default `\001`, the
same as hive's default delimiter
|
@@ -181,7 +182,7 @@ The result of this example matching is:
### file_format_type [string]
File type, supported as the following file types:
-`text` `csv` `parquet` `orc` `json` `excel` `xml` `binary` `markdown`
+`text` `csv` `parquet` `orc` `json` `excel` `xml` `binary` `markdown` `pdf`
If you assign file type to `json`, you should also assign schema option to
tell connector how to parse data to the row you want.
For example:
upstream data is the following:
@@ -251,7 +252,7 @@ at the same time.
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:
+Each extracted element is converted to a document-element 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)
@@ -272,6 +273,17 @@ The option defaults to `false`, so the original Markdown
schema is unchanged unl
Note: Markdown format only supports reading, not writing.
+If you assign file type to `pdf`, SeaTunnel can parse PDF files and extract
structured document elements.
+PDF uses the same document-element row schema described above.
+
+The main PDF-specific behaviors are:
+
+- **With outline**: Extracts `heading`, `paragraph`, `image`, and `link`
elements. Headings are derived from the outline structure, and elements are
organized into a parent-child hierarchy reflecting the document's logical
structure.
+- **Without outline**: Extracts only `paragraph` and `image` elements in a
flat structure without hierarchy.
+- `element_type` values for PDF are `heading`, `paragraph`, `image`, and
`link`.
+
+Note: Only single-column (top-to-bottom) PDF layouts are supported.
Multi-column layouts (e.g., side-by-side two-column documents) are not
supported and may produce incorrect text ordering.
+
### compress_codec [string]
The compress codec of files and the details that supported as the following
shown:
diff --git a/docs/zh/connectors/source/CosFile.md
b/docs/zh/connectors/source/CosFile.md
index a1d401c353..3f81b513ae 100644
--- a/docs/zh/connectors/source/CosFile.md
+++ b/docs/zh/connectors/source/CosFile.md
@@ -35,6 +35,7 @@ import ChangeLog from '../changelog/connector-file-cos.md';
- [x] xml
- [x] binary
- [x] markdown
+ - [x] pdf
## 描述
@@ -95,7 +96,7 @@ import ChangeLog from '../changelog/connector-file-cos.md';
文件类型，支持以下文件类型：
-`text` `csv` `parquet` `orc` `json` `excel` `xml` `binary` `markdown`
+`text` `csv` `parquet` `orc` `json` `excel` `xml` `binary` `markdown` `pdf`
如果您将文件类型设置为“json”，您还应该分配模式选项，告诉连接器如何将数据解析到所需的行。
@@ -184,7 +185,7 @@ schema {
如果您将文件类型指定为 `markdown`，SeaTunnel 可以解析 markdown 文件并提取结构化数据。
markdown 解析器提取各种元素，包括标题、段落、列表、代码块、表格等。
-每个元素都转换为具有以下架构的行：
+每个提取出的元素都会转换为一条文档元素结构化记录，schema 如下：
- `element_id`：元素的唯一标识符
- `element_type`：元素类型（Heading、Paragraph、ListItem 等）
- `heading_level`：标题级别（1-6，非标题元素为 null）
@@ -204,6 +205,18 @@ markdown 解析器提取各种元素，包括标题、段落、列表、代码
该选项默认值为 `false`，因此只有显式启用后才会改变原始 Markdown schema。
注意：Markdown 格式仅支持读取，不支持写入。
+
+如果您将文件类型指定为 `pdf`，SeaTunnel 可以解析 PDF 文件并提取结构化的文档元素。
+PDF 使用与上文相同的文档元素 schema。
+
+PDF 特有的解析行为如下：
+
+- **有大纲**：提取 `heading`（标题）、`paragraph`（段落）、`image`（图片）和
`link`（链接）元素。标题从大纲结构中派生，元素按照文档的逻辑结构组织为父子层级关系。
+- **无大纲**：仅提取 `paragraph`（段落）和 `image`（图片）元素，以扁平结构呈现，不包含层级关系。
+- `element_type` 在 PDF 场景下可能为 `heading`、`paragraph`、`image` 或 `link`。
+
+注意：仅支持单栏（从上到下）PDF 布局。不支持多栏布局（例如并排的双栏文档），可能会产生不正确的文本顺序。
+
根据此要求，您需要确保源端和目标端使用“二进制”格式进行文件同步同时。您可以在下面的示例中找到具体用法。
### bucket [string]
diff --git a/docs/zh/connectors/source/FtpFile.md
b/docs/zh/connectors/source/FtpFile.md
index eaa0925cb0..74c6c81af9 100644
--- a/docs/zh/connectors/source/FtpFile.md
+++ b/docs/zh/connectors/source/FtpFile.md
@@ -30,6 +30,8 @@ import ChangeLog from '../changelog/connector-file-ftp.md';
- [x] Excel
- [x] XML
- [x] 二进制
+ - [x] markdown
+ - [x] pdf
## 描述
@@ -181,7 +183,7 @@ abc.*
文件类型，支持以下文件类型：
-`text` `csv` `parquet` `orc` `json` `excel` `xml` `binary` `markdown`
+`text` `csv` `parquet` `orc` `json` `excel` `xml` `binary` `markdown` `pdf`
如果您将文件类型指定为 `json`，您还需要指定 schema 选项以告诉连接器如何将数据解析为您所需的行。
@@ -261,7 +263,7 @@ schema {
如果您将文件类型指定为 `markdown`，SeaTunnel 可以解析 markdown 文件并提取结构化数据。
markdown 解析器提取各种元素，包括标题、段落、列表、代码块、表格等。
-每个元素都转换为具有以下架构的行：
+每个提取出的元素都会转换为一条文档元素结构化记录，schema 如下：
- `element_id`：元素的唯一标识符
- `element_type`：元素类型（Heading、Paragraph、ListItem 等）
- `heading_level`：标题级别（1-6，非标题元素为 null）
@@ -282,6 +284,17 @@ markdown 解析器提取各种元素，包括标题、段落、列表、代码
注意：Markdown 格式仅支持读取，不支持写入。
+如果您将文件类型指定为 `pdf`，SeaTunnel 可以解析 PDF 文件并提取结构化的文档元素。
+PDF 使用与上文相同的文档元素 schema。
+
+PDF 特有的解析行为如下：
+
+- **有大纲**：提取 `heading`（标题）、`paragraph`（段落）、`image`（图片）和
`link`（链接）元素。标题从大纲结构中派生，元素按照文档的逻辑结构组织为父子层级关系。
+- **无大纲**：仅提取 `paragraph`（段落）和 `image`（图片）元素，以扁平结构呈现，不包含层级关系。
+- `element_type` 在 PDF 场景下可能为 `heading`、`paragraph`、`image` 或 `link`。
+
+注意：仅支持单栏（从上到下）PDF 布局。不支持多栏布局（例如并排的双栏文档），可能会产生不正确的文本顺序。
+
### connection_mode [string]
目标 FTP 连接模式，默认为主动模式，支持以下模式：
diff --git a/docs/zh/connectors/source/HdfsFile.md
b/docs/zh/connectors/source/HdfsFile.md
index 04da70a457..f6a9d1126b 100644
--- a/docs/zh/connectors/source/HdfsFile.md
+++ b/docs/zh/connectors/source/HdfsFile.md
@@ -36,6 +36,7 @@ import ChangeLog from '../changelog/connector-file-hadoop.md';
- [x] xml
- [x] binary
- [x] markdown
+ - [x] pdf
## 描述
@@ -53,7 +54,7 @@ import ChangeLog from '../changelog/connector-file-hadoop.md';
|----------------------------|---------|------|---------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| path | string | 是 | - | 源文件路径。
|
| tables_configs | list | 否 | - | 在一个
Source 块中配置多张 HDFS 源表。每一项都使用与单表 `HdfsFile` Source 相同的参数，并可通过 `schema.table`
设置传递给下游的表名。 |
-| file_format_type | string | 是 | - |
我们支持以下文件类型：`text` `csv` `parquet` `orc` `json` `excel` `xml` `binary`
`markdown`。请注意，最终文件名将以文件格式的后缀结束，文本文件的后缀是 `txt`。
|
+| file_format_type | string | 是 | - |
我们支持以下文件类型：`text` `csv` `parquet` `orc` `json` `excel` `xml` `binary`
`markdown` `pdf`。请注意，最终文件名将以文件格式的后缀结束，文本文件的后缀是 `txt`。
|
| fs.defaultFS | string | 是 | - | 以
`hdfs://` 开头的 hadoop 集群地址，例如：`hdfs://hadoopcluster`
|
| read_columns | list | 否 | - |
数据源的读取列列表，用户可以使用它来实现字段投影。支持列投影的文件类型如下所示：[text,json,csv,orc,parquet,excel,xml]。提示：如果用户想在读取
`text` `json` `csv` 文件时使用此功能，必须配置 schema 选项。
|
| hdfs_site_path | string | 否 | - |
`hdfs-site.xml` 的路径，用于加载 namenodes 的 ha 配置
|
@@ -104,11 +105,11 @@ import ChangeLog from
'../changelog/connector-file-hadoop.md';
文件类型，支持以下文件类型：
-`text` `csv` `parquet` `orc` `json` `excel` `xml` `binary` `markdown`
+`text` `csv` `parquet` `orc` `json` `excel` `xml` `binary` `markdown` `pdf`
如果您将文件类型指定为 `markdown`，SeaTunnel 可以解析 markdown 文件并提取结构化数据。
markdown 解析器提取各种元素，包括标题、段落、列表、代码块、表格等。
-每个元素都转换为具有以下架构的行：
+每个提取出的元素都会转换为一条文档元素结构化记录，schema 如下：
- `element_id`：元素的唯一标识符
- `element_type`：元素类型（Heading、Paragraph、ListItem 等）
- `heading_level`：标题级别（1-6，非标题元素为 null）
@@ -129,6 +130,17 @@ markdown 解析器提取各种元素，包括标题、段落、列表、代码
注意：Markdown 格式仅支持读取，不支持写入。
+如果您将文件类型指定为 `pdf`，SeaTunnel 可以解析 PDF 文件并提取结构化的文档元素。
+PDF 使用与上文相同的文档元素 schema。
+
+PDF 特有的解析行为如下：
+
+- **有大纲**：提取 `heading`（标题）、`paragraph`（段落）、`image`（图片）和
`link`（链接）元素。标题从大纲结构中派生，元素按照文档的逻辑结构组织为父子层级关系。
+- **无大纲**：仅提取 `paragraph`（段落）和 `image`（图片）元素，以扁平结构呈现，不包含层级关系。
+- `element_type` 在 PDF 场景下可能为 `heading`、`paragraph`、`image` 或 `link`。
+
+注意：仅支持单栏（从上到下）PDF 布局。不支持多栏布局（例如并排的双栏文档），可能会产生不正确的文本顺序。
+
### tables_configs [list]
当一个 HDFS Source 需要读取多张表或多个目录时，可以使用 `tables_configs`。每一项都可以单独配置
diff --git a/docs/zh/connectors/source/Hive.md
b/docs/zh/connectors/source/Hive.md
index 2cbadc7408..fb59b04f11 100644
--- a/docs/zh/connectors/source/Hive.md
+++ b/docs/zh/connectors/source/Hive.md
@@ -8,7 +8,7 @@ import ChangeLog from '../changelog/connector-hive.md';
从 Hive 读取数据。
-使用 markdown 格式时，SeaTunnel 可以解析存储在 Hive 表中的 markdown
文件并提取结构化数据，包括标题、段落、列表、代码块和表格等元素。每个元素都转换为具有以下架构的行：
+使用 markdown 格式时，SeaTunnel 可以解析存储在 Hive 表中的 markdown
文件并提取结构化数据，包括标题、段落、列表、代码块和表格等元素。每个提取出的元素都会转换为一条文档元素结构化记录，schema 如下：
- `element_id`：元素的唯一标识符
- `element_type`：元素类型（Heading、Paragraph、ListItem 等）
- `heading_level`：标题级别（1-6，非标题元素为 null）
diff --git a/docs/zh/connectors/source/LocalFile.md
b/docs/zh/connectors/source/LocalFile.md
index 4017e67e82..ca1031c3a4 100644
--- a/docs/zh/connectors/source/LocalFile.md
+++ b/docs/zh/connectors/source/LocalFile.md
@@ -35,6 +35,7 @@ import ChangeLog from '../changelog/connector-file-local.md';
- [x] xml
- [x] binary
- [x] markdown
+ - [x] pdf
## 描述
@@ -104,7 +105,7 @@ import ChangeLog from
'../changelog/connector-file-local.md';
文件类型，支持以下文件类型：
-`text` `csv` `parquet` `orc` `json` `excel` `xml` `binary` `markdown`
+`text` `csv` `parquet` `orc` `json` `excel` `xml` `binary` `markdown` `pdf`
如果您将文件类型指定为 `json`，您还应该指定 schema 选项来告诉连接器如何将数据解析为您想要的行。
@@ -195,7 +196,7 @@ schema {
如果您将文件类型指定为 `markdown`，SeaTunnel 可以解析 markdown 文件并提取结构化数据。
markdown 解析器提取各种元素，包括标题、段落、列表、代码块、表格等。
-每个元素都转换为具有以下架构的行：
+每个提取出的元素都会转换为一条文档元素结构化记录，schema 如下：
- `element_id`：元素的唯一标识符
- `element_type`：元素类型（Heading、Paragraph、ListItem 等）
- `heading_level`：标题级别（1-6，非标题元素为 null）
@@ -216,6 +217,17 @@ markdown 解析器提取各种元素，包括标题、段落、列表、代码
注意：Markdown 格式仅支持读取，不支持写入。
+如果您将文件类型指定为 `pdf`，SeaTunnel 可以解析 PDF 文件并提取结构化的文档元素。
+PDF 使用与上文相同的文档元素 schema。
+
+PDF 特有的解析行为如下：
+
+- **有大纲**：提取 `heading`（标题）、`paragraph`（段落）、`image`（图片）和
`link`（链接）元素。标题从大纲结构中派生，元素按照文档的逻辑结构组织为父子层级关系。
+- **无大纲**：仅提取 `paragraph`（段落）和 `image`（图片）元素，以扁平结构呈现，不包含层级关系。
+- `element_type` 在 PDF 场景下可能为 `heading`、`paragraph`、`image` 或 `link`。
+
+注意：仅支持单栏（从上到下）PDF 布局。不支持多栏布局（例如并排的双栏文档），可能会产生不正确的文本顺序。
+
### read_columns [list]
数据源的读取列列表，用户可以使用它来实现字段投影。
@@ -651,6 +663,31 @@ LocalFile {
```
+### 读取 PDF 文件
+
+```hocon
+
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ LocalFile {
+ path = "/data/documents/"
+ file_format_type = "pdf"
+ }
+}
+
+sink {
+ Console {
+ }
+}
+
+```
+
+为获得最佳效果，请使用包含大纲（书签/目录）的 PDF 文件。这使解析器能够提取具有层级信息的标题。
+
### 传输二进制文件
```hocon
diff --git a/docs/zh/connectors/source/ObsFile.md
b/docs/zh/connectors/source/ObsFile.md
index 90f155711a..c914b3daa9 100644
--- a/docs/zh/connectors/source/ObsFile.md
+++ b/docs/zh/connectors/source/ObsFile.md
@@ -35,6 +35,7 @@ import ChangeLog from '../changelog/connector-file-obs.md';
- [x] json
- [x] excel
- [x] markdown
+ - [x] pdf
## 描述
@@ -65,7 +66,7 @@ import ChangeLog from '../changelog/connector-file-obs.md';
| 参数名 | 类型 | 必须 | 默认值 | 描述
|
|---------------------------|---------|----|---------------------|-----------------------------------------|
| path | string | 是 | - | 目标目录路径
|
-| file_format_type | string | 是 | - | 文件类型
|
+| file_format_type | string | 是 | - |
文件类型，支持以下文件类型：`text` `csv` `parquet` `orc` `json` `excel` `xml` `binary`
`markdown` `pdf` |
| bucket | string | 是 | - | OBS
文件系统的桶地址，例如：`obs://obs-bucket-name` |
| access_key | string | 是 | - | OBS
文件系统的访问密钥 |
| access_secret | string | 是 | - | OBS
文件系统的访问密钥 |
@@ -112,6 +113,17 @@ markdown 解析器提取各种元素，包括标题、段落、列表、代码
注意：Markdown 格式仅支持读取，不支持写入。
+如果您将文件类型指定为 `pdf`，SeaTunnel 可以解析 PDF 文件并提取结构化的文档元素。
+PDF 使用与上文相同的文档元素 schema。
+
+PDF 特有的解析行为如下：
+
+- **有大纲**：提取 `heading`（标题）、`paragraph`（段落）、`image`（图片）和
`link`（链接）元素。标题从大纲结构中派生，元素按照文档的逻辑结构组织为父子层级关系。
+- **无大纲**：仅提取 `paragraph`（段落）和 `image`（图片）元素，以扁平结构呈现，不包含层级关系。
+- `element_type` 在 PDF 场景下可能为 `heading`、`paragraph`、`image` 或 `link`。
+
+注意：仅支持单栏（从上到下）PDF 布局。不支持多栏布局（例如并排的双栏文档），可能会产生不正确的文本顺序。
+
### sort_files_by_modification_time [boolean]
是否按修改时间降序排序文件。默认值为 `false`。
diff --git a/docs/zh/connectors/source/OssFile.md
b/docs/zh/connectors/source/OssFile.md
index b2018d9cb1..57e389dfdc 100644
--- a/docs/zh/connectors/source/OssFile.md
+++ b/docs/zh/connectors/source/OssFile.md
@@ -46,6 +46,7 @@ import ChangeLog from '../changelog/connector-file-oss.md';
- [x] xml
- [x] binary
- [x] markdown
+ - [x] pdf
## 数据类型映射
@@ -186,7 +187,7 @@ schema {
| 名称 | 类型 | 是否必需 | 默认值 | 描述
|
|----------------------------|---------|------|--------------------|------------------------------------------------------------------------------------------------------------------------------------------------------|
| path | string | 是 | - |
需要读取的Oss路径，可以有子路径，但子路径需要满足一定的格式要求。具体要求可以参考"parse_partition_from_path"选项
|
-| file_format_type | string | 是 | - |
文件类型，支持以下文件类型：`text` `csv` `parquet` `orc` `json` `excel` `xml` `binary`
`markdown` |
+| file_format_type | string | 是 | - |
文件类型，支持以下文件类型：`text` `csv` `parquet` `orc` `json` `excel` `xml` `binary`
`markdown` `pdf`
|
| bucket | string | 是 | - |
oss文件系统的bucket地址，例如：`oss://seatunnel-test`。
|
| endpoint | string | 是 | - | fs oss端点
|
| read_columns | list | 否 | - |
数据源的读取列列表，用户可以使用它来实现字段投影。支持列投影的文件类型如下所示：`text` `csv` `parquet` `orc` `json`
`excel` `xml`。如果用户想在读取`text` `json` `csv`文件时使用此功能，必须配置"schema"选项。 |
@@ -251,11 +252,11 @@ schema {
文件类型，支持以下文件类型：
-`text` `csv` `parquet` `orc` `json` `excel` `xml` `binary` `markdown`
+`text` `csv` `parquet` `orc` `json` `excel` `xml` `binary` `markdown` `pdf`
如果您将文件类型指定为 `markdown`，SeaTunnel 可以解析 markdown 文件并提取结构化数据。
markdown 解析器提取各种元素，包括标题、段落、列表、代码块、表格等。
-每个元素都转换为具有以下架构的行：
+每个提取出的元素都会转换为一条文档元素结构化记录，schema 如下：
- `element_id`：元素的唯一标识符
- `element_type`：元素类型（Heading、Paragraph、ListItem 等）
- `heading_level`：标题级别（1-6，非标题元素为 null）
@@ -276,6 +277,17 @@ markdown 解析器提取各种元素，包括标题、段落、列表、代码
注意：Markdown 格式仅支持读取，不支持写入。
+如果您将文件类型指定为 `pdf`，SeaTunnel 可以解析 PDF 文件并提取结构化的文档元素。
+PDF 使用与上文相同的文档元素 schema。
+
+PDF 特有的解析行为如下：
+
+- **有大纲**：提取 `heading`（标题）、`paragraph`（段落）、`image`（图片）和
`link`（链接）元素。标题从大纲结构中派生，元素按照文档的逻辑结构组织为父子层级关系。
+- **无大纲**：仅提取 `paragraph`（段落）和 `image`（图片）元素，以扁平结构呈现，不包含层级关系。
+- `element_type` 在 PDF 场景下可能为 `heading`、`paragraph`、`image` 或 `link`。
+
+注意：仅支持单栏（从上到下）PDF 布局。不支持多栏布局（例如并排的双栏文档），可能会产生不正确的文本顺序。
+
### quote_char [string]
用于包裹 CSV 字段的单字符，可保证包含逗号、换行符或引号的字段被正确解析。
diff --git a/docs/zh/connectors/source/OssJindoFile.md
b/docs/zh/connectors/source/OssJindoFile.md
index a7559992fc..e873b535df 100644
--- a/docs/zh/connectors/source/OssJindoFile.md
+++ b/docs/zh/connectors/source/OssJindoFile.md
@@ -35,6 +35,7 @@ import ChangeLog from
'../changelog/connector-file-oss-jindo.md';
- [x] xml
- [x] binary
- [x] markdown
+ - [x] pdf
## 描述
@@ -59,7 +60,7 @@ import ChangeLog from
'../changelog/connector-file-oss-jindo.md';
| 参数名 | 类型 | 必须 | 默认值 | 描述
|
|---------------------------|---------|----|-----------------------------|-------------------------------------------------------------------------------|
| path | string | 是 | - |
目标目录路径 |
-| file_format_type | string | 是 | - |
文件类型 |
+| file_format_type | string | 是 | - |
文件类型，支持以下文件类型：`text` `csv` `parquet` `orc` `json` `excel` `xml` `binary`
`markdown` `pdf`
|
| bucket | string | 是 | - | OSS
文件系统的桶地址 |
| access_key | string | 是 | - | OSS
文件系统的访问密钥 |
| access_secret | string | 是 | - | OSS
文件系统的访问密钥 |
@@ -119,6 +120,18 @@ markdown 解析器提取各种元素，包括标题、段落、列表、代码
注意：Markdown 格式仅支持读取，不支持写入。
+如果您将文件类型指定为 `pdf`，SeaTunnel 可以解析 PDF 文件并提取结构化的文档元素。
+PDF 使用与上文相同的文档元素 schema。
+
+PDF 特有的解析行为如下：
+
+- **有大纲**：提取 `heading`（标题）、`paragraph`（段落）、`image`（图片）和
`link`（链接）元素。标题从大纲结构中派生，元素按照文档的逻辑结构组织为父子层级关系。
+- **无大纲**：仅提取 `paragraph`（段落）和 `image`（图片）元素，以扁平结构呈现，不包含层级关系。
+- `element_type` 在 PDF 场景下可能为 `heading`、`paragraph`、`image` 或 `link`。
+
+注意：仅支持单栏（从上到下）PDF 布局。不支持多栏布局（例如并排的双栏文档），可能会产生不正确的文本顺序。
+
+
## 变更日志
<ChangeLog />
diff --git a/docs/zh/connectors/source/S3File.md
b/docs/zh/connectors/source/S3File.md
index e91b7a0758..9e5768fefa 100644
--- a/docs/zh/connectors/source/S3File.md
+++ b/docs/zh/connectors/source/S3File.md
@@ -35,6 +35,7 @@ import ChangeLog from '../changelog/connector-file-s3.md';
- [x] xml
- [x] binary
- [x] markdown
+ - [x] pdf
## 描述
@@ -192,7 +193,7 @@ schema {
| 名称 | 类型 | 是否必需 | 默认值
| 描述
|
|---------------------------------|---------|------|-------------------------------------------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| path | string | 是 | -
|
需要读取的s3路径，可以有子路径，但子路径需要满足一定的格式要求。具体要求可以参考"parse_partition_from_path"选项
|
-| file_format_type | string | 是 | -
| 文件类型，支持以下文件类型：`text` `csv` `parquet` `orc` `json`
`excel` `xml` `binary` `markdown`
|
+| file_format_type | string | 是 | -
| 文件类型，支持以下文件类型：`text` `csv` `parquet` `orc` `json`
`excel` `xml` `binary` `markdown` `pdf`
|
| bucket | string | 是 | -
|
s3文件系统的bucket地址，例如：`s3n://seatunnel-test`，如果您使用`s3a`协议，此参数应为`s3a://seatunnel-test`。
|
| fs.s3a.endpoint | string | 是 | -
| fs s3a端点
|
| fs.s3a.aws.credentials.provider | string | 是 |
com.amazonaws.auth.InstanceProfileCredentialsProvider |
s3a的认证方式。我们目前只支持`org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider`和`com.amazonaws.auth.InstanceProfileCredentialsProvider`。有关凭据提供程序的更多信息，您可以查看[Hadoop
AWS文档](https://hadoop.apache.org/docs/stable/hadoop-aws/tools/hadoop-aws/index.html#Simple_name.2Fsecret_credentials_with_SimpleAWSCredentialsProvider.2A)
|
@@ -384,11 +385,11 @@ abc.*
文件类型，支持以下文件类型：
-`text` `csv` `parquet` `orc` `json` `excel` `xml` `binary` `markdown`
+`text` `csv` `parquet` `orc` `json` `excel` `xml` `binary` `markdown` `pdf`
如果您将文件类型指定为 `markdown`，SeaTunnel 可以解析 markdown 文件并提取结构化数据。
markdown 解析器提取各种元素，包括标题、段落、列表、代码块、表格等。
-每个元素都转换为具有以下架构的行：
+每个提取出的元素都会转换为一条文档元素结构化记录，schema 如下：
- `element_id`：元素的唯一标识符
- `element_type`：元素类型（Heading、Paragraph、ListItem 等）
- `heading_level`：标题级别（1-6，非标题元素为 null）
@@ -409,6 +410,17 @@ markdown 解析器提取各种元素，包括标题、段落、列表、代码
注意：Markdown 格式仅支持读取，不支持写入。
+如果您将文件类型指定为 `pdf`，SeaTunnel 可以解析 PDF 文件并提取结构化的文档元素。
+PDF 使用与上文相同的文档元素 schema。
+
+PDF 特有的解析行为如下：
+
+- **有大纲**：提取 `heading`（标题）、`paragraph`（段落）、`image`（图片）和
`link`（链接）元素。标题从大纲结构中派生，元素按照文档的逻辑结构组织为父子层级关系。
+- **无大纲**：仅提取 `paragraph`（段落）和 `image`（图片）元素，以扁平结构呈现，不包含层级关系。
+- `element_type` 在 PDF 场景下可能为 `heading`、`paragraph`、`image` 或 `link`。
+
+注意：仅支持单栏（从上到下）PDF 布局。不支持多栏布局（例如并排的双栏文档），可能会产生不正确的文本顺序。
+
### schema [config]
仅在文件格式类型为 text、json、excel、xml 或 csv（或其他无法从元数据中读取 schema 的格式）时需要配置。
diff --git a/docs/zh/connectors/source/SftpFile.md
b/docs/zh/connectors/source/SftpFile.md
index db3dce2ca7..555427c1c5 100644
--- a/docs/zh/connectors/source/SftpFile.md
+++ b/docs/zh/connectors/source/SftpFile.md
@@ -30,6 +30,7 @@ import ChangeLog from '../changelog/connector-file-sftp.md';
- [x] xml
- [x] binary
- [x] markdown
+ - [x] pdf
## 描述
@@ -181,7 +182,7 @@ abc.*
### file_format_type [string]
文件类型，支持以下文件类型：
-`text` `csv` `parquet` `orc` `json` `excel` `xml` `binary` `markdown`
+`text` `csv` `parquet` `orc` `json` `excel` `xml` `binary` `markdown` `pdf`
如果您将文件类型指定为`json`，您还应该指定schema选项来告诉连接器如何将数据解析为您想要的行。
例如：
上游数据如下：
@@ -249,7 +250,7 @@ schema {
如果您将文件类型指定为 `markdown`，SeaTunnel 可以解析 markdown 文件并提取结构化数据。
markdown 解析器提取各种元素，包括标题、段落、列表、代码块、表格等。
-每个元素都转换为具有以下架构的行：
+每个提取出的元素都会转换为一条文档元素结构化记录，schema 如下：
- `element_id`：元素的唯一标识符
- `element_type`：元素类型（Heading、Paragraph、ListItem 等）
- `heading_level`：标题级别（1-6，非标题元素为 null）
@@ -269,6 +270,18 @@ markdown 解析器提取各种元素，包括标题、段落、列表、代码
该选项默认值为 `false`，因此只有显式启用后才会改变原始 Markdown schema。
注意：Markdown 格式仅支持读取，不支持写入。
+
+如果您将文件类型指定为 `pdf`，SeaTunnel 可以解析 PDF 文件并提取结构化的文档元素。
+PDF 使用与上文相同的文档元素 schema。
+
+PDF 特有的解析行为如下：
+
+- **有大纲**：提取 `heading`（标题）、`paragraph`（段落）、`image`（图片）和
`link`（链接）元素。标题从大纲结构中派生，元素按照文档的逻辑结构组织为父子层级关系。
+- **无大纲**：仅提取 `paragraph`（段落）和 `image`（图片）元素，以扁平结构呈现，不包含层级关系。
+- `element_type` 在 PDF 场景下可能为 `heading`、`paragraph`、`image` 或 `link`。
+
+注意：仅支持单栏（从上到下）PDF 布局。不支持多栏布局（例如并排的双栏文档），可能会产生不正确的文本顺序。
+
在此要求下，您需要确保源和接收器同时使用`binary`格式进行文件同步。
### compress_codec [string]
diff --git
a/seatunnel-ci-tools/src/test/java/org/apache/seatunnel/api/file/AllFileSpecificationCheckTest.java
b/seatunnel-ci-tools/src/test/java/org/apache/seatunnel/api/file/AllFileSpecificationCheckTest.java
index c8af7b4b6d..52549ef8ae 100644
---
a/seatunnel-ci-tools/src/test/java/org/apache/seatunnel/api/file/AllFileSpecificationCheckTest.java
+++
b/seatunnel-ci-tools/src/test/java/org/apache/seatunnel/api/file/AllFileSpecificationCheckTest.java
@@ -47,7 +47,8 @@ public class AllFileSpecificationCheckTest {
public static void beforeAll() throws IOException {
List<String> fileTypesCanNotRead =
Arrays.asList(
- "parquet", "orc", "xlsx", "xls", "png", "jar", "lzo",
"zip", "ico", "jks");
+ "parquet", "orc", "xlsx", "xls", "png", "jar", "lzo",
"zip", "ico", "jks",
+ "pdf");
List<String> fileCanNotRead =
Arrays.asList(
"seatunnel-connectors-v2/connector-file/connector-file-base/src/test/resources/encoding/gbk.json",
diff --git a/seatunnel-connectors-v2/connector-file/connector-file-base/pom.xml
b/seatunnel-connectors-v2/connector-file/connector-file-base/pom.xml
index 8742d0a47c..d49f238785 100644
--- a/seatunnel-connectors-v2/connector-file/connector-file-base/pom.xml
+++ b/seatunnel-connectors-v2/connector-file/connector-file-base/pom.xml
@@ -42,6 +42,7 @@
<easyexcel.version>4.0.3</easyexcel.version>
<fastexcel-reader.version>0.18.4</fastexcel-reader.version>
<flexmark-all.version>0.62.2</flexmark-all.version>
+ <pdfbox.version>3.0.2</pdfbox.version>
</properties>
<dependencyManagement>
@@ -192,6 +193,12 @@
<version>${mockito.version}</version>
<scope>test</scope>
</dependency>
+
+ <dependency>
+ <groupId>org.apache.pdfbox</groupId>
+ <artifactId>pdfbox</artifactId>
+ <version>${pdfbox.version}</version>
+ </dependency>
</dependencies>
<build>
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/config/DocumentElement.java
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/config/DocumentElement.java
new file mode 100644
index 0000000000..b4700af7a4
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/config/DocumentElement.java
@@ -0,0 +1,193 @@
+/*
+ * 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.config;
+
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+
+/**
+ * Document element structure, used to represent various elements (titles,
paragraphs, etc.) in a
+ * document
+ */
+public class DocumentElement {
+
+ /** Element Unique Identifier */
+ private String elementId;
+
+ /** Element types: heading, paragraph, link, image, etc. */
+ private String elementType;
+
+ /** Title level, only valid when elementType is heading */
+ private Integer headingLevel;
+
+ /** Element text content */
+ private String text;
+
+ /** Page numbers, page numbers in PDF files, starting from 1 */
+ private Integer pageNumber;
+
+ /** Position index in the document, starting from 0 */
+ private Integer positionIndex;
+
+ /** Parent element ID, used to construct the hierarchy */
+ private String parentId;
+
+ /** List of child Element ids */
+ private List<String> childIds;
+
+ public DocumentElement() {
+ this.elementId = "uuid-elem-" + UUID.randomUUID().toString();
+ this.childIds = new ArrayList<>();
+ }
+
+ public DocumentElement(String elementType, String text) {
+ this();
+ this.elementType = elementType;
+ this.text = text;
+ }
+
+ public DocumentElement(
+ String elementType, String text, Integer pageNumber, Integer
positionIndex) {
+ this(elementType, text);
+ this.pageNumber = pageNumber;
+ this.positionIndex = positionIndex;
+ }
+
+ // Getters and Setters
+ public String getElementId() {
+ return elementId;
+ }
+
+ public void setElementId(String elementId) {
+ this.elementId = elementId;
+ }
+
+ public String getElementType() {
+ return elementType;
+ }
+
+ public void setElementType(String elementType) {
+ this.elementType = elementType;
+ }
+
+ public Integer getHeadingLevel() {
+ return headingLevel;
+ }
+
+ public void setHeadingLevel(Integer headingLevel) {
+ this.headingLevel = headingLevel;
+ }
+
+ public String getText() {
+ return text;
+ }
+
+ public void setText(String text) {
+ this.text = text;
+ }
+
+ public Integer getPageNumber() {
+ return pageNumber;
+ }
+
+ public void setPageNumber(Integer pageNumber) {
+ this.pageNumber = pageNumber;
+ }
+
+ public Integer getPositionIndex() {
+ return positionIndex;
+ }
+
+ public void setPositionIndex(Integer positionIndex) {
+ this.positionIndex = positionIndex;
+ }
+
+ public String getParentId() {
+ return parentId;
+ }
+
+ public void setParentId(String parentId) {
+ this.parentId = parentId;
+ }
+
+ public List<String> getChildIds() {
+ return childIds;
+ }
+
+ public void setChildIds(List<String> childIds) {
+ this.childIds = childIds;
+ }
+
+ /** Add the child element ID */
+ public void addChildId(String childId) {
+ if (this.childIds == null) {
+ this.childIds = new ArrayList<>();
+ }
+ this.childIds.add(childId);
+ }
+
+ /** Remove the child element ID */
+ public void removeChildId(String childId) {
+ if (this.childIds != null) {
+ this.childIds.remove(childId);
+ }
+ }
+
+ @Override
+ public String toString() {
+ return "DocumentElement{"
+ + "elementId='"
+ + elementId
+ + '\''
+ + ", elementType='"
+ + elementType
+ + '\''
+ + ", headingLevel="
+ + headingLevel
+ + ", text='"
+ + (text != null && text.length() > 50 ? text.substring(0, 50)
+ "..." : text)
+ + '\''
+ + ", pageNumber="
+ + pageNumber
+ + ", positionIndex="
+ + positionIndex
+ + ", parentId='"
+ + parentId
+ + '\''
+ + ", childIds="
+ + childIds
+ + '}';
+ }
+
+ public SeaTunnelRow toSeaTunnelRow() {
+ return new SeaTunnelRow(
+ new Object[] {
+ elementId,
+ elementType,
+ headingLevel,
+ text,
+ pageNumber,
+ positionIndex,
+ parentId,
+ childIds == null || childIds.isEmpty() ? null :
childIds.toArray(new String[0])
+ });
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/config/FileFormat.java
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/config/FileFormat.java
index 1253b82725..82fc91db8c 100644
---
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/config/FileFormat.java
+++
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/config/FileFormat.java
@@ -37,6 +37,7 @@ import
org.apache.seatunnel.connectors.seatunnel.file.source.reader.JsonReadStra
import
org.apache.seatunnel.connectors.seatunnel.file.source.reader.MarkdownReadStrategy;
import
org.apache.seatunnel.connectors.seatunnel.file.source.reader.OrcReadStrategy;
import
org.apache.seatunnel.connectors.seatunnel.file.source.reader.ParquetReadStrategy;
+import
org.apache.seatunnel.connectors.seatunnel.file.source.reader.PdfReadStrategy;
import
org.apache.seatunnel.connectors.seatunnel.file.source.reader.ReadStrategy;
import
org.apache.seatunnel.connectors.seatunnel.file.source.reader.TextReadStrategy;
import
org.apache.seatunnel.connectors.seatunnel.file.source.reader.XmlReadStrategy;
@@ -184,7 +185,17 @@ public enum FileFormat implements Serializable {
return new MarkdownReadStrategy();
}
},
- ;
+ PDF("pdf") {
+ @Override
+ public WriteStrategy getWriteStrategy(FileSinkConfig fileSinkConfig) {
+ throw new UnsupportedOperationException("File format 'pdf' does
not support writing.");
+ }
+
+ @Override
+ public ReadStrategy getReadStrategy() {
+ return new PdfReadStrategy();
+ }
+ };
private final String[] suffix;
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/PdfReadStrategy.java
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/PdfReadStrategy.java
new file mode 100644
index 0000000000..479eaf4af8
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/PdfReadStrategy.java
@@ -0,0 +1,1034 @@
+/*
+ * 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.shade.org.apache.commons.lang3.StringUtils;
+
+import org.apache.seatunnel.api.source.Collector;
+import org.apache.seatunnel.api.table.type.ArrayType;
+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.seatunnel.connectors.seatunnel.file.config.DocumentElement;
+import
org.apache.seatunnel.connectors.seatunnel.file.exception.FileConnectorErrorCode;
+import
org.apache.seatunnel.connectors.seatunnel.file.exception.FileConnectorException;
+
+import org.apache.pdfbox.Loader;
+import org.apache.pdfbox.contentstream.PDFStreamEngine;
+import org.apache.pdfbox.contentstream.operator.Operator;
+import org.apache.pdfbox.cos.COSBase;
+import org.apache.pdfbox.cos.COSName;
+import org.apache.pdfbox.pdmodel.PDDocument;
+import org.apache.pdfbox.pdmodel.PDDocumentCatalog;
+import org.apache.pdfbox.pdmodel.PDPage;
+import org.apache.pdfbox.pdmodel.common.PDRectangle;
+import org.apache.pdfbox.pdmodel.graphics.PDXObject;
+import org.apache.pdfbox.pdmodel.graphics.form.PDFormXObject;
+import org.apache.pdfbox.pdmodel.graphics.image.PDImageXObject;
+import org.apache.pdfbox.pdmodel.interactive.action.PDAction;
+import org.apache.pdfbox.pdmodel.interactive.action.PDActionGoTo;
+import org.apache.pdfbox.pdmodel.interactive.action.PDActionURI;
+import org.apache.pdfbox.pdmodel.interactive.annotation.PDAnnotation;
+import org.apache.pdfbox.pdmodel.interactive.annotation.PDAnnotationLink;
+import
org.apache.pdfbox.pdmodel.interactive.documentnavigation.destination.PDDestination;
+import
org.apache.pdfbox.pdmodel.interactive.documentnavigation.destination.PDNamedDestination;
+import
org.apache.pdfbox.pdmodel.interactive.documentnavigation.destination.PDPageDestination;
+import
org.apache.pdfbox.pdmodel.interactive.documentnavigation.destination.PDPageXYZDestination;
+import
org.apache.pdfbox.pdmodel.interactive.documentnavigation.outline.PDDocumentOutline;
+import
org.apache.pdfbox.pdmodel.interactive.documentnavigation.outline.PDOutlineItem;
+import org.apache.pdfbox.text.PDFTextStripper;
+import org.apache.pdfbox.text.PDFTextStripperByArea;
+import org.apache.pdfbox.text.TextPosition;
+import org.apache.pdfbox.util.Matrix;
+
+import lombok.Getter;
+import lombok.extern.slf4j.Slf4j;
+
+import java.awt.geom.Rectangle2D;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardCopyOption;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+/**
+ * PDF reading strategy V2 for RAG scenarios Extracts structured content
including chapter titles,
+ * paragraph text, and images
+ */
+@Slf4j
+public class PdfReadStrategy extends AbstractReadStrategy {
+
+ @Override
+ public void read(String path, String tableId, Collector<SeaTunnelRow>
output)
+ throws IOException, FileConnectorException {
+
+ Path tempPdfPath = null;
+ try {
+ tempPdfPath = createTempPdfPath();
+ try (InputStream inputStream =
hadoopFileSystemProxy.getInputStream(path)) {
+ Files.copy(inputStream, tempPdfPath,
StandardCopyOption.REPLACE_EXISTING);
+ }
+
+ try (PDDocument document = Loader.loadPDF(tempPdfPath.toFile())) {
+ List<DocumentElement> elements =
extractPdfDocumentElements(document);
+
+ log.info(
+ "PDF file '{}' processed successfully, generated {}
elements",
+ path,
+ elements.size());
+
+ for (DocumentElement element : elements) {
+ output.collect(element.toSeaTunnelRow());
+ }
+ }
+ } catch (Exception e) {
+ throw new FileConnectorException(
+ FileConnectorErrorCode.FILE_READ_FAILED,
+ String.format("Failed to parse PDF document [%s]: %s",
path, e.getMessage()),
+ e);
+ } finally {
+ if (tempPdfPath != null) {
+ deleteTempPdfPath(tempPdfPath);
+ }
+ }
+ }
+
+ Path createTempPdfPath() throws IOException {
+ return Files.createTempFile("seatunnel-pdf-read-", ".pdf");
+ }
+
+ void deleteTempPdfPath(Path tempPdfPath) throws IOException {
+ Files.deleteIfExists(tempPdfPath);
+ }
+
+ @Override
+ public SeaTunnelRowType getSeaTunnelRowTypeInfo(String path) throws
FileConnectorException {
+ return new SeaTunnelRowType(
+ new String[] {
+ "element_id",
+ "element_type",
+ "heading_level",
+ "text",
+ "page_number",
+ "position_index",
+ "parent_id",
+ "child_ids"
+ },
+ new SeaTunnelDataType[] {
+ BasicType.STRING_TYPE,
+ BasicType.STRING_TYPE,
+ BasicType.INT_TYPE,
+ BasicType.STRING_TYPE,
+ BasicType.INT_TYPE,
+ BasicType.INT_TYPE,
+ BasicType.STRING_TYPE,
+ ArrayType.STRING_ARRAY_TYPE
+ });
+ }
+
+ /** Extract all document elements including headings, paragraphs, and
images */
+ private List<DocumentElement> extractPdfDocumentElements(PDDocument
document)
+ throws IOException {
+ List<DocumentElement> elements = new ArrayList<>();
+
+ PDDocumentCatalog catalog = document.getDocumentCatalog();
+ PDDocumentOutline documentOutline = catalog.getDocumentOutline();
+
+ // Check if outline exists
+ if (documentOutline != null && documentOutline.getFirstChild() !=
null) {
+ List<CoordinateInfo> headingCoordinates = new ArrayList<>();
+
+ // Extract headings from outline
+ List<DocumentElement> headings =
+ extractHeadingsFromOutline(document, documentOutline,
headingCoordinates);
+
+ // Build coordinate lookup map for O(1) access
+ Map<String, CoordinateInfo> coordMap =
+ headingCoordinates.stream()
+ .filter(c -> c.getElementId() != null)
+ .collect(
+ Collectors.toMap(
+ CoordinateInfo::getElementId, c ->
c, (a, b) -> a));
+
+ // Extract paragraphs and associate with headings
+ List<DocumentElement> paragraphs =
+ extractParagraphsForHeadings(document, headings, coordMap);
+
+ // Extract images and associate with headings
+ List<DocumentElement> images = extractImages(document,
headingCoordinates);
+
+ List<DocumentElement> links = extractLinks(document,
headingCoordinates);
+
+ // Merge headings and paragraphs maintaining hierarchy
+ elements.addAll(mergeElements(headings, paragraphs, images,
links));
+ } else {
+ // No outline available, treat all content as paragraphs
+ log.info("No outline found in PDF, processing only paragraphs and
images elements");
+ elements.addAll(extractParagraphsAndImages(document));
+ }
+
+ return elements;
+ }
+
+ /** Extract headings from PDF outline structure with coordinate
information */
+ private List<DocumentElement> extractHeadingsFromOutline(
+ PDDocument document, PDDocumentOutline outline,
List<CoordinateInfo> headingCoordinates)
+ throws IOException {
+ List<DocumentElement> headings = new ArrayList<>();
+
+ PDOutlineItem current = outline.getFirstChild();
+ int positionIndex = 0;
+
+ while (current != null) {
+ positionIndex =
+ extractOutlineItemRecursively(
+ document,
+ current,
+ null,
+ 1,
+ positionIndex,
+ headings,
+ headingCoordinates);
+ current = current.getNextSibling();
+ }
+
+ log.debug("Extracted {} headings from outline with coordinates",
headings.size());
+ return headings;
+ }
+
+ /** Recursively extract outline items and build heading hierarchy with
coordinates */
+ private int extractOutlineItemRecursively(
+ PDDocument document,
+ PDOutlineItem item,
+ String parentId,
+ int level,
+ int positionIndex,
+ List<DocumentElement> headings,
+ List<CoordinateInfo> coordinates)
+ throws IOException {
+
+ DocumentElement heading =
+ convertOutlineToHeading(document, item, parentId, level,
positionIndex);
+ if (heading == null) {
+ return positionIndex;
+ }
+
+ headings.add(heading);
+ positionIndex++;
+
+ // Extract coordinate information
+ CoordinateInfo coord = extractCoordinateFromOutline(document, item,
heading.getElementId());
+ if (coord != null) {
+ coordinates.add(coord);
+ }
+
+ // Process children
+ PDOutlineItem child = item.getFirstChild();
+ while (child != null) {
+ int startIndex = positionIndex;
+ positionIndex =
+ extractOutlineItemRecursively(
+ document,
+ child,
+ heading.getElementId(),
+ level + 1,
+ positionIndex,
+ headings,
+ coordinates);
+
+ // Update parent's child_ids with direct children only
+ if (positionIndex > startIndex) {
+ for (int i = startIndex; i < positionIndex && i <
headings.size(); i++) {
+ DocumentElement childElement = headings.get(i);
+ if
(heading.getElementId().equals(childElement.getParentId())) {
+ heading.addChildId(childElement.getElementId());
+ break;
+ }
+ }
+ }
+
+ child = child.getNextSibling();
+ }
+
+ return positionIndex;
+ }
+
+ /** Convert PDF outline item to DocumentElement heading */
+ private DocumentElement convertOutlineToHeading(
+ PDDocument document, PDOutlineItem item, String parentId, int
level, int positionIndex)
+ throws IOException {
+
+ if (item == null || StringUtils.isBlank(item.getTitle())) {
+ return null;
+ }
+
+ String title = item.getTitle().trim();
+ DocumentElement heading = new DocumentElement("heading", title);
+ heading.setHeadingLevel(Math.min(level, 6)); // Limit to max level 6
+ heading.setPositionIndex(positionIndex);
+ heading.setParentId(parentId);
+
+ // Get page number from destination
+ try {
+ PDPage destinationPage = item.findDestinationPage(document);
+ if (destinationPage != null) {
+ int pageIndex = document.getPages().indexOf(destinationPage);
+ if (pageIndex >= 0) {
+ heading.setPageNumber(pageIndex + 1);
+ }
+ }
+ } catch (IOException e) {
+ log.warn("Could not determine page number for heading: {}", title);
+ }
+
+ return heading;
+ }
+
+ /** Extract coordinate information from PDF outline item */
+ private CoordinateInfo extractCoordinateFromOutline(
+ PDDocument document, PDOutlineItem item, String elementId) {
+ try {
+ PDDestination destination = item.getDestination();
+
+ if (destination == null) {
+ // Check for action-based destinations
+ PDAction action = item.getAction();
+ if (action instanceof PDActionGoTo) {
+ destination = ((PDActionGoTo) action).getDestination();
+ }
+ }
+
+ PDPageDestination pageDestination = null;
+ if (destination instanceof PDNamedDestination) {
+ pageDestination =
+ document.getDocumentCatalog()
+ .findNamedDestinationPage((PDNamedDestination)
destination);
+ if (pageDestination == null) {
+ return null;
+ }
+
+ } else if (destination instanceof PDPageDestination) {
+ pageDestination = (PDPageDestination) destination;
+ }
+
+ if (pageDestination instanceof PDPageXYZDestination) {
+ PDPageXYZDestination xyzDest = (PDPageXYZDestination)
pageDestination;
+
+ // Get page number
+ int pageNumber = xyzDest.retrievePageNumber();
+ if (pageNumber < 0) {
+ PDPage page = xyzDest.getPage();
+ if (page != null) {
+ pageNumber = document.getPages().indexOf(page);
+ }
+ }
+
+ // Get coordinates - note that PDF coordinates have origin at
bottom-left
+ float x = xyzDest.getLeft();
+ float y = xyzDest.getTop();
+
+ return new CoordinateInfo(pageNumber, x, y, elementId);
+ }
+
+ } catch (IOException e) {
+ log.warn("Could not extract coordinates for outline item: {}",
item.getTitle());
+ }
+
+ return null;
+ }
+
+ /** Extract paragraphs for each heading using coordinate-based approach */
+ private List<DocumentElement> extractParagraphsForHeadings(
+ PDDocument document,
+ List<DocumentElement> headings,
+ Map<String, CoordinateInfo> coordMap)
+ throws IOException {
+
+ List<DocumentElement> paragraphs = new ArrayList<>();
+
+ for (int i = 0; i < headings.size(); i++) {
+ DocumentElement heading = headings.get(i);
+ DocumentElement nextHeading = (i + 1 < headings.size()) ?
headings.get(i + 1) : null;
+
+ String paragraphText =
+ extractParagraphUsingCoordinates(document, heading,
nextHeading, coordMap);
+
+ paragraphText = cleanParagraphText(paragraphText);
+
+ if (StringUtils.isNotBlank(paragraphText)) {
+ DocumentElement paragraph = new DocumentElement("paragraph",
paragraphText.trim());
+ paragraph.setParentId(heading.getElementId());
+ paragraph.setPageNumber(heading.getPageNumber());
+ paragraph.setHeadingLevel(null);
+ paragraph.setChildIds(new ArrayList<>());
+ paragraphs.add(paragraph);
+ }
+ }
+
+ return paragraphs;
+ }
+
+ /** Extract paragraph text using coordinate information */
+ private String extractParagraphUsingCoordinates(
+ PDDocument document,
+ DocumentElement heading,
+ DocumentElement nextHeading,
+ Map<String, CoordinateInfo> coordMap)
+ throws IOException {
+
+ CoordinateInfo currentCoord = coordMap.get(heading.getElementId());
+ CoordinateInfo nextCoord =
+ (nextHeading != null) ?
coordMap.get(nextHeading.getElementId()) : null;
+
+ if (currentCoord != null) {
+ return extractTextUsingCoordinates(document, currentCoord,
nextCoord);
+ } else {
+ log.info(
+ "No coordinates found for heading: {}, falling back to
text-based extraction",
+ heading.getText());
+ List<String> pageContents = extractPageContents(document);
+ return extractParagraphTextForHeading(heading, nextHeading,
pageContents);
+ }
+ }
+
+ /** Extract text using coordinate information */
+ private String extractTextUsingCoordinates(
+ PDDocument document, CoordinateInfo startCoord, CoordinateInfo
endCoord)
+ throws IOException {
+
+ int startPageIndex = startCoord.getPageNumber();
+ int endPageIndex =
+ endCoord != null ? endCoord.getPageNumber() :
document.getNumberOfPages() - 1;
+
+ float startCoordY = startCoord.getY();
+ float endCoordY = endCoord == null ? 0 : endCoord.getY();
+
+ // Obtain the Y-axis coordinate at the beginning of the paragraph
+ ParagraphCoordinateExtractor paragraphExtractor =
+ new ParagraphCoordinateExtractor(document);
+ float paragraphEndY = paragraphExtractor.getParagraphEndY(startCoord,
startPageIndex + 1);
+
+ if (paragraphEndY != 0.0f) {
+ startCoordY = paragraphEndY;
+ }
+
+ PDFTextStripperByArea stripper = new PDFTextStripperByArea();
+ stripper.setSortByPosition(true);
+
+ StringBuilder content = new StringBuilder();
+
+ float regionStartY, regionHeight;
+
+ for (int pageIndex = startPageIndex; pageIndex <= endPageIndex;
pageIndex++) {
+ PDPage page = document.getPage(pageIndex);
+ PDRectangle mediaBox = page.getMediaBox();
+
+ if (pageIndex == startPageIndex && pageIndex == endPageIndex) {
+ regionStartY = mediaBox.getHeight() - startCoordY;
+ regionHeight = startCoordY - endCoordY;
+ } else {
+ if (pageIndex == startPageIndex) {
+ regionStartY = mediaBox.getHeight() - startCoordY;
+ regionHeight = startCoordY;
+ } else if (pageIndex == endPageIndex) {
+ regionStartY = 0;
+ regionHeight = mediaBox.getHeight() - endCoordY;
+ } else {
+ regionStartY = 0;
+ regionHeight = mediaBox.getHeight();
+ }
+ }
+
+ // Create an extraction area
+ Rectangle2D region =
+ new Rectangle2D.Float(0, regionStartY,
mediaBox.getWidth(), regionHeight);
+
+ stripper.addRegion("page" + pageIndex, region);
+ stripper.extractRegions(page);
+
+ String pageContent = stripper.getTextForRegion("page" + pageIndex);
+ if (pageContent != null && !pageContent.trim().isEmpty()) {
+ content.append(pageContent).append("\n");
+ }
+ }
+
+ return content.toString().trim();
+ }
+
+ /** Extract page contents as text */
+ private List<String> extractPageContents(PDDocument document) throws
IOException {
+ List<String> pageContents = new ArrayList<>();
+ PDFTextStripper stripper = new PDFTextStripper();
+ stripper.setSortByPosition(true);
+
+ for (int i = 1; i <= document.getNumberOfPages(); i++) {
+ stripper.setStartPage(i);
+ stripper.setEndPage(i);
+ String pageText = stripper.getText(document);
+
+ // Remove header/footer noise
+ pageContents.add(pageText);
+ }
+
+ return pageContents;
+ }
+
+ /** Extract paragraph text for a specific heading */
+ private String extractParagraphTextForHeading(
+ DocumentElement heading, DocumentElement nextHeading, List<String>
pageContents) {
+
+ if (heading.getPageNumber() == null || heading.getPageNumber() < 1) {
+ return "";
+ }
+
+ String headingTitle = heading.getText();
+ String nextTitle = (nextHeading != null) ? nextHeading.getText() :
null;
+ Integer nextPageNum = (nextHeading != null) ?
nextHeading.getPageNumber() : null;
+
+ return extractTextBetweenHeadings(
+ heading.getPageNumber(), nextPageNum, headingTitle, nextTitle,
pageContents);
+ }
+
+ /** This method finds the actual content between the current heading and
the next heading */
+ private String extractTextBetweenHeadings(
+ int startPage,
+ Integer endPage,
+ String startTitle,
+ String endTitle,
+ List<String> pageContents) {
+
+ StringBuilder content = new StringBuilder();
+ boolean startCollecting = false;
+ boolean foundStartTitle = false;
+
+ int maxPage = (endPage != null) ? endPage : pageContents.size();
+
+ // Iterate through pages to find content between headings
+ for (int page = startPage; page <= maxPage; page++) {
+ String pageText = pageContents.get(page - 1);
+
+ if (StringUtils.isBlank(pageText)) {
+ continue;
+ }
+
+ String[] lines = pageText.split("\\n");
+
+ for (String line : lines) {
+ String trimmedLine = line.trim();
+
+ if (StringUtils.isBlank(trimmedLine)) {
+ continue;
+ }
+
+ // Check if this line contains the starting heading
+ if (!foundStartTitle
+ && StringUtils.isNotBlank(startTitle)
+ && lineContainsHeading(trimmedLine, startTitle)) {
+
+ // Skip the heading line itself
+ foundStartTitle = true;
+ startCollecting = true;
+ continue;
+ }
+
+ // Check if this line contains the ending heading
+ if (foundStartTitle
+ && StringUtils.isNotBlank(endTitle)
+ && lineContainsHeading(trimmedLine, endTitle)) {
+
+ // Stop collecting when we hit the next heading
+ startCollecting = false;
+ break;
+ }
+
+ // Collect content between headings
+ if (startCollecting) {
+ content.append(trimmedLine).append("\n");
+ }
+ }
+ }
+
+ return content.toString().trim();
+ }
+
+ /** Check if a line contains a specific heading */
+ private boolean lineContainsHeading(String line, String heading) {
+ if (StringUtils.isBlank(line) || StringUtils.isBlank(heading)) {
+ return false;
+ }
+
+ if (line.equals(heading)) {
+ return true;
+ }
+
+ // Check if line ends with the heading.
+ // The reason for using the endsWith method is that
+ // for the original title: "1.1 Introduction", the extract head title
will only take the
+ // value of "Introduction"
+ if (line.endsWith(heading)) {
+ if (line.length() == heading.length()) {
+ return true;
+ }
+ char charBefore = line.charAt(line.length() - heading.length() -
1);
+ return Character.isWhitespace(charBefore)
+ || Character.isDigit(charBefore)
+ || charBefore == '.'
+ || charBefore == ')';
+ }
+
+ return false;
+ }
+
+ /** Process document with no outline - only contains paragraphs and images
element. */
+ private List<DocumentElement> extractParagraphsAndImages(PDDocument
document)
+ throws IOException {
+ List<DocumentElement> elements = new ArrayList<>();
+ List<String> pageContents = extractPageContents(document);
+ List<DocumentElement> images = extractImages(document, new
ArrayList<>());
+
+ for (int i = 0; i < pageContents.size(); i++) {
+ String pageText = pageContents.get(i);
+ if (StringUtils.isNotBlank(pageText)) {
+ DocumentElement paragraph = new DocumentElement("paragraph",
pageText.trim());
+ paragraph.setPageNumber(i + 1);
+ paragraph.setPositionIndex(i);
+ elements.add(paragraph);
+ }
+ }
+
+ elements.addAll(images);
+
+ // sorted by pageNumber
+ elements.sort(Comparator.comparingInt(DocumentElement::getPageNumber));
+
+ return elements;
+ }
+
+ List<DocumentElement> extractImages(
+ PDDocument document, List<CoordinateInfo> headingCoordinates)
throws IOException {
+
+ PdfImageExtractor pdfImageExtractor = new PdfImageExtractor(document);
+
+ for (int pageNum = 0; pageNum < document.getNumberOfPages();
pageNum++) {
+ pdfImageExtractor.processPage(document.getPage(pageNum));
+ }
+
+ List<CoordinateInfo> imagesCoordinates =
pdfImageExtractor.getImagesCoordinates();
+
+ // When no headings exist (no-outline PDF), create flat image elements
directly
+ if (headingCoordinates.isEmpty()) {
+ return createFlatImageElements(imagesCoordinates);
+ }
+
+ return assignElementsToHeadings(
+ document,
+ imagesCoordinates,
+ headingCoordinates,
+ "image",
+ coord ->
+ String.format(
+ "image_page_%d_pos_(%.0f,%.0f)",
+ coord.getPageNumber() + 1, coord.getX(),
coord.getY()));
+ }
+
+ /** Create flat image elements for PDFs without outline structure */
+ private List<DocumentElement> createFlatImageElements(List<CoordinateInfo>
imagesCoordinates) {
+ List<DocumentElement> elements = new ArrayList<>();
+ for (int i = 0; i < imagesCoordinates.size(); i++) {
+ CoordinateInfo coord = imagesCoordinates.get(i);
+ DocumentElement element = new DocumentElement();
+ element.setElementType("image");
+ element.setText(
+ String.format(
+ "image_page_%d_pos_(%.0f,%.0f)",
+ coord.getPageNumber() + 1, coord.getX(),
coord.getY()));
+ element.setPageNumber(coord.getPageNumber() + 1);
+ element.setPositionIndex(i);
+ element.setParentId(null);
+ element.setChildIds(new ArrayList<>());
+ elements.add(element);
+ }
+ return elements;
+ }
+
+ /** Determine if element belongs to a specific heading based on position
rules */
+ private boolean isElementBelongsToHead(
+ CoordinateInfo elementCoordinate, CoordinateInfo currentHead,
CoordinateInfo nextHead) {
+
+ int elementPage = elementCoordinate.getPageNumber();
+ float elementY = elementCoordinate.getY();
+
+ int currentPage = currentHead.getPageNumber();
+ float currentY = currentHead.getY();
+
+ // Case 1: element is before the first heading
+ if (elementPage < currentPage || (elementPage == currentPage &&
elementY > currentY)) {
+ return false;
+ }
+
+ // Case 2: No next heading - element belongs to current heading if
it's after current
+ // heading
+ if (nextHead == null) {
+ return elementPage > currentPage || (elementPage == currentPage &&
elementY < currentY);
+ }
+
+ int nextPage = nextHead.getPageNumber();
+ float nextY = nextHead.getY();
+
+ // Case 3: element page is between current and next heading pages
+ if (elementPage > currentPage && elementPage < nextPage) {
+ return true;
+ }
+
+ // Case 4: element is on same page as current heading
+ if (elementPage == currentPage) {
+ // element must be below current heading
+ if (elementY >= currentY) {
+ return false;
+ }
+
+ // If next heading is on same page, element must be above next
heading
+ if (nextPage == currentPage) {
+ return elementY > nextY;
+ }
+
+ // Next heading is on different page, element belongs to current
heading
+ return true;
+ }
+
+ // Case 5: element is on same page as next heading
+ if (elementPage == nextPage) {
+ // element must be above (higher Y) next heading to belong to
current heading
+ return elementY > nextY;
+ }
+
+ return false;
+ }
+
+ /** Obtain all the link elements in the document */
+ public List<DocumentElement> extractLinks(
+ PDDocument document, List<CoordinateInfo> headingCoordinates)
throws IOException {
+
+ List<CoordinateInfo> linksCoordinateInfo = new ArrayList<>();
+
+ for (int pageNum = 0; pageNum < document.getNumberOfPages();
pageNum++) {
+ PDPage page = document.getPage(pageNum);
+ for (PDAnnotation annot : page.getAnnotations()) {
+ if (annot instanceof PDAnnotationLink) {
+ PDAnnotationLink link = (PDAnnotationLink) annot;
+ PDRectangle rectangle = link.getRectangle();
+ PDAction action = link.getAction();
+ if (action instanceof PDActionURI) {
+ PDActionURI uri = (PDActionURI) action;
+ linksCoordinateInfo.add(
+ new CoordinateInfo(
+ pageNum,
+ rectangle.getUpperRightX(),
+ rectangle.getUpperRightY(),
+ null,
+ uri.getURI()));
+ }
+ }
+ }
+ }
+
+ return assignElementsToHeadings(
+ document, linksCoordinateInfo, headingCoordinates, "link",
CoordinateInfo::getText);
+ }
+
+ private List<DocumentElement> assignElementsToHeadings(
+ PDDocument document,
+ List<CoordinateInfo> elementCoordinates,
+ List<CoordinateInfo> headingCoordinates,
+ String elementType,
+ java.util.function.Function<CoordinateInfo, String> textExtractor)
{
+
+ List<DocumentElement> elements = new ArrayList<>();
+
+ for (int i = 0; i < headingCoordinates.size(); i++) {
+ CoordinateInfo heading = headingCoordinates.get(i);
+ CoordinateInfo nextHeading =
+ (i + 1 < headingCoordinates.size()) ?
headingCoordinates.get(i + 1) : null;
+ int endPage =
+ nextHeading == null
+ ? document.getNumberOfPages()
+ : nextHeading.getPageNumber() + 1;
+
+ IntStream.range(heading.getPageNumber(), endPage)
+ .forEach(
+ pageIndex ->
+ elementCoordinates.stream()
+ .filter(coord ->
coord.getPageNumber() == pageIndex)
+ .filter(
+ coord ->
+
isElementBelongsToHead(
+ coord,
heading, nextHeading))
+ .forEach(
+ coord -> {
+ DocumentElement
element =
+ new
DocumentElement();
+
element.setParentId(heading.getElementId());
+ element.setPageNumber(
+
coord.getPageNumber() + 1);
+
element.setChildIds(new ArrayList<>());
+
element.setElementType(elementType);
+ String text =
textExtractor.apply(coord);
+ if (text != null) {
+
element.setText(text);
+ }
+ elements.add(element);
+ }));
+ }
+
+ return elements;
+ }
+
+ /** Merge headings, paragraphs and images element */
+ private List<DocumentElement> mergeElements(
+ List<DocumentElement> headings,
+ List<DocumentElement> paragraphs,
+ List<DocumentElement> images,
+ List<DocumentElement> links) {
+
+ List<DocumentElement> mergeResult = new ArrayList<>();
+
+ Map<String, DocumentElement> paragraphMap =
+ paragraphs.stream()
+ .collect(
+ Collectors.toMap(
+ DocumentElement::getParentId,
+ Function.identity(),
+ (existing, replacement) -> existing));
+
+ Map<String, List<DocumentElement>> imageMap =
+ images.stream()
+ .filter(element -> element.getParentId() != null)
+
.collect(Collectors.groupingBy(DocumentElement::getParentId));
+
+ Map<String, List<DocumentElement>> linkMap =
+ links.stream()
+ .filter(element -> element.getParentId() != null)
+
.collect(Collectors.groupingBy(DocumentElement::getParentId));
+
+ for (DocumentElement heading : headings) {
+ String parentId = heading.getElementId();
+
+ mergeResult.add(heading);
+
+ if (paragraphMap.containsKey(parentId)) {
+ DocumentElement paragraph = paragraphMap.get(parentId);
+ List<String> childIds = heading.getChildIds();
+ childIds.add(paragraph.getElementId());
+ heading.setChildIds(childIds);
+
+ mergeResult.add(paragraph);
+ } else {
+ log.debug(
+ "No paragraph text found under heading '{}', which is
expected for headings without body content.",
+ heading.getText());
+ }
+
+ if (imageMap.containsKey(parentId)) {
+ List<DocumentElement> imageChild = imageMap.get(parentId);
+
+ imageChild.forEach(
+ imageElement -> {
+
heading.getChildIds().add(imageElement.getElementId());
+ mergeResult.add(imageElement);
+ });
+ }
+
+ if (linkMap.containsKey(parentId)) {
+ List<DocumentElement> linkChild = linkMap.get(parentId);
+
+ linkChild.forEach(
+ linkElement -> {
+
heading.getChildIds().add(linkElement.getElementId());
+ mergeResult.add(linkElement);
+ });
+ }
+ }
+
+ for (int i = 0; i < mergeResult.size(); i++) {
+ mergeResult.get(i).setPositionIndex(i);
+ }
+
+ return mergeResult;
+ }
+
+ private String cleanParagraphText(String paragraphText) {
+ return paragraphText.replace("\r\n", "\n");
+ }
+
+ /** Coordinate information for PDF elements */
+ private static class CoordinateInfo {
+ private final int pageNumber;
+ private final float x;
+ private final float y;
+ private final String elementId;
+ private final String text;
+
+ public CoordinateInfo(int pageNumber, float x, float y, String
elementId, String text) {
+ this.pageNumber = pageNumber;
+ this.x = x;
+ this.y = y;
+ this.elementId = elementId;
+ this.text = text;
+ }
+
+ public CoordinateInfo(int pageNumber, float x, float y, String
elementId) {
+ this(pageNumber, x, y, elementId, null);
+ }
+
+ public int getPageNumber() {
+ return pageNumber;
+ }
+
+ public float getX() {
+ return x;
+ }
+
+ public float getY() {
+ return y;
+ }
+
+ public String getElementId() {
+ return elementId;
+ }
+
+ public String getText() {
+ return text;
+ }
+ }
+
+ /** Finds the paragraph boundary below an outline heading based on text
position changes. */
+ private static class ParagraphCoordinateExtractor {
+ private final PDDocument document;
+
+ public ParagraphCoordinateExtractor(PDDocument document) {
+ this.document = document;
+ }
+
+ public float getParagraphEndY(CoordinateInfo outlineItemCoordinate,
int pageNum)
+ throws IOException {
+
+ float[] lastTextFontSize = {0.0f};
+ String[] lastTextFontName = {""};
+
+ float[] paragraphEndY = {0.0f};
+
+ PDFTextStripper textStripper =
+ new PDFTextStripper() {
+ @Override
+ protected void writeString(String text,
List<TextPosition> textPositions)
+ throws IOException {
+
+ if (!textPositions.isEmpty()) {
+ TextPosition textPosition =
textPositions.get(0);
+
+ if (textPosition.getEndY() <=
outlineItemCoordinate.getY()) {
+
+ if (paragraphEndY[0] > 0.0f) {
+ return;
+ }
+
+ if
(lastTextFontName[0].equals(StringUtils.EMPTY)) {
+ lastTextFontName[0] =
textPosition.getFont().getName();
+ }
+
+ if (textPosition.getFontSize() <
lastTextFontSize[0]
+ || !textPosition
+ .getFont()
+ .getName()
+
.equals(lastTextFontName[0])) {
+ paragraphEndY[0] =
textPosition.getEndY();
+ return;
+ } else {
+ lastTextFontSize[0] =
textPosition.getFontSize();
+ }
+ }
+ }
+
+ super.writeString(text, textPositions);
+ }
+ };
+
+ textStripper.setStartPage(pageNum);
+ textStripper.setEndPage(pageNum);
+ textStripper.getText(document);
+
+ return paragraphEndY[0];
+ }
+ }
+
+ /** Extracts image coordinates from PDF drawing operations for later
heading assignment. */
+ private static class PdfImageExtractor extends PDFStreamEngine {
+ private final PDDocument document;
+ @Getter private List<CoordinateInfo> imagesCoordinates = new
ArrayList<>();
+
+ /** @throws IOException If there is an error loading text stripper
properties. */
+ public PdfImageExtractor(PDDocument document) throws IOException {
+ this.document = document;
+ }
+
+ /**
+ * @param operator The operation to perform.
+ * @param operands The list of arguments.
+ * @throws IOException If there is an error processing the operation.
+ */
+ @Override
+ protected void processOperator(Operator operator, List<COSBase>
operands)
+ throws IOException {
+ PDPage currentPage = getCurrentPage();
+ int pageNumber = document.getPages().indexOf(currentPage);
+
+ String operation = operator.getName();
+ if ("Do".equals(operation)) {
+ COSName objectName = (COSName) operands.get(0);
+ // get the PDF object
+ PDXObject xobject = getResources().getXObject(objectName);
+ // check if the object is an image object
+ if (xobject instanceof PDImageXObject) {
+ Matrix ctmNew =
getGraphicsState().getCurrentTransformationMatrix();
+
+ CoordinateInfo coordinateInfo =
+ new CoordinateInfo(
+ pageNumber,
+ ctmNew.getTranslateX(),
+ ctmNew.getTranslateY(),
+ null);
+
+ imagesCoordinates.add(coordinateInfo);
+ } else if (xobject instanceof PDFormXObject) {
+ PDFormXObject form = (PDFormXObject) xobject;
+ showForm(form);
+ }
+ } else {
+ super.processOperator(operator, operands);
+ }
+ }
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/PdfReadStrategyTest.java
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/PdfReadStrategyTest.java
new file mode 100644
index 0000000000..bb7b7442be
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/source/reader/PdfReadStrategyTest.java
@@ -0,0 +1,348 @@
+/*
+ * 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.table.type.SeaTunnelRow;
+import org.apache.seatunnel.connectors.seatunnel.file.config.DocumentElement;
+import org.apache.seatunnel.connectors.seatunnel.file.config.HadoopConf;
+import
org.apache.seatunnel.connectors.seatunnel.file.exception.FileConnectorException;
+
+import org.apache.pdfbox.pdmodel.PDDocument;
+import org.apache.pdfbox.pdmodel.PDPage;
+import org.apache.pdfbox.pdmodel.PDPageContentStream;
+import org.apache.pdfbox.pdmodel.common.PDRectangle;
+import org.apache.pdfbox.pdmodel.font.PDType1Font;
+import org.apache.pdfbox.pdmodel.font.Standard14Fonts;
+import org.apache.pdfbox.pdmodel.graphics.image.PDImageXObject;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import lombok.extern.slf4j.Slf4j;
+
+import javax.imageio.ImageIO;
+
+import java.awt.image.BufferedImage;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.net.URISyntaxException;
+import java.net.URL;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static
org.apache.hadoop.fs.CommonConfigurationKeysPublic.FS_DEFAULT_NAME_DEFAULT;
+
+@Slf4j
+public class PdfReadStrategyTest {
+
+ @Test
+ public void testReadPdfStrategy()
+ throws URISyntaxException, IOException, FileConnectorException {
+ URL resource =
this.getClass().getResource("/pdf_read_strategy_test.pdf");
+
+ String path = Paths.get(resource.toURI()).toString();
+ PdfReadStrategy pdfReadStrategy = new PdfReadStrategy();
+ LocalConf localConf = new LocalConf(FS_DEFAULT_NAME_DEFAULT);
+ pdfReadStrategy.init(localConf);
+ TempCollector tempCollector = new TempCollector();
+ pdfReadStrategy.read(path, "", tempCollector);
+
+ List<SeaTunnelRow> rows = tempCollector.getRows();
+
+ List<SeaTunnelRow> headingElements = getHeadingElements(rows);
+
+ // verify heading elements count
+ Assertions.assertEquals(11, headingElements.size());
+
+ Assertions.assertEquals("heading", rows.get(0).getField(1));
+ Assertions.assertEquals(1, rows.get(0).getField(2));
+ Assertions.assertEquals(
+ "The Essential Guide to Groceries: Shopping, Storing, and
Enjoying Food at Home",
+ rows.get(0).getField(3));
+ Assertions.assertEquals(1, rows.get(0).getField(4));
+ Assertions.assertEquals(0, rows.get(0).getField(5));
+ Assertions.assertNull(rows.get(0).getField(6));
+
+ // child_ids must be a String[] array (not a List or comma-separated
String)
+ String[] childIds = (String[]) rows.get(0).getField(7);
+ Assertions.assertNotNull(childIds);
+ Assertions.assertEquals(11, childIds.length);
+
+ // check paragraph
+ String expectedParagraph =
+ "Groceries play a vital role in daily life, touching every
aspect of health, convenience, and enjoyment.\n"
+ + "This comprehensive guide covers all things
groceries—from what to shop for, strategies to save money, storage tips, and
even how groceries have changed in the\n"
+ + "modern era.";
+
+ Assertions.assertEquals(
+ expectedParagraph.length(),
String.valueOf(rows.get(1).getField(3)).length());
+ Assertions.assertEquals(expectedParagraph, rows.get(1).getField(3));
+
+ // check link element type
+ int lastIndex = rows.size() - 1;
+ Assertions.assertEquals("link", rows.get(lastIndex).getField(1));
+ // check link element text
+ Assertions.assertEquals("https://example.com";,
rows.get(lastIndex).getField(3));
+ // check link element page
+ Assertions.assertEquals(3, rows.get(lastIndex).getField(4));
+ }
+
+ @Test
+ public void testReadInvalidPdfThrowsException() {
+ PdfReadStrategy strategy = new PdfReadStrategy();
+ LocalConf localConf = new LocalConf(FS_DEFAULT_NAME_DEFAULT);
+ strategy.init(localConf);
+ TempCollector collector = new TempCollector();
+ Assertions.assertThrows(
+ FileConnectorException.class,
+ () -> strategy.read("/nonexistent/path/file.pdf", "",
collector));
+ }
+
+ @Test
+ public void testChildIdsIsArrayType() throws URISyntaxException,
IOException {
+ URL resource =
this.getClass().getResource("/pdf_read_strategy_test.pdf");
+ String path = Paths.get(resource.toURI()).toString();
+ PdfReadStrategy pdfReadStrategy = new PdfReadStrategy();
+ LocalConf localConf = new LocalConf(FS_DEFAULT_NAME_DEFAULT);
+ pdfReadStrategy.init(localConf);
+ TempCollector tempCollector = new TempCollector();
+ pdfReadStrategy.read(path, "", tempCollector);
+
+ for (SeaTunnelRow row : tempCollector.getRows()) {
+ Object childIdsField = row.getField(7);
+ Assertions.assertTrue(
+ childIdsField == null || childIdsField instanceof String[],
+ "child_ids must be null or String[], got: "
+ + (childIdsField == null ? "null" :
childIdsField.getClass()));
+ }
+ }
+
+ @Test
+ public void testImageElementsHaveText() throws URISyntaxException,
IOException {
+ URL resource =
this.getClass().getResource("/pdf_read_strategy_test.pdf");
+ String path = Paths.get(resource.toURI()).toString();
+ PdfReadStrategy pdfReadStrategy = new PdfReadStrategy();
+ LocalConf localConf = new LocalConf(FS_DEFAULT_NAME_DEFAULT);
+ pdfReadStrategy.init(localConf);
+ TempCollector tempCollector = new TempCollector();
+ pdfReadStrategy.read(path, "", tempCollector);
+
+ List<SeaTunnelRow> imageRows =
+ tempCollector.getRows().stream()
+ .filter(row -> "image".equals(row.getField(1)))
+ .collect(Collectors.toList());
+
+ for (SeaTunnelRow imageRow : imageRows) {
+ String text = (String) imageRow.getField(3);
+ Assertions.assertNotNull(text, "image element must have text");
+ Assertions.assertTrue(
+ text.startsWith("image_page_"), "image text must start
with 'image_page_'");
+ }
+ }
+
+ @Test
+ public void testNoOutlinePdfWithImagesExtractsImageElements() throws
IOException {
+ Path tempPdf = createNoOutlinePdfWithImage();
+ try {
+ PdfReadStrategy pdfReadStrategy = new PdfReadStrategy();
+ LocalConf localConf = new LocalConf(FS_DEFAULT_NAME_DEFAULT);
+ pdfReadStrategy.init(localConf);
+ TempCollector tempCollector = new TempCollector();
+ pdfReadStrategy.read(tempPdf.toString(), "", tempCollector);
+
+ List<SeaTunnelRow> rows = tempCollector.getRows();
+ Assertions.assertFalse(rows.isEmpty(), "Should produce at least
one element");
+
+ List<SeaTunnelRow> paragraphs =
+ rows.stream()
+ .filter(row -> "paragraph".equals(row.getField(1)))
+ .collect(Collectors.toList());
+ Assertions.assertFalse(paragraphs.isEmpty(), "Should have
paragraph elements");
+
+ List<SeaTunnelRow> images =
+ rows.stream()
+ .filter(row -> "image".equals(row.getField(1)))
+ .collect(Collectors.toList());
+ Assertions.assertFalse(
+ images.isEmpty(), "Should have image elements for
no-outline PDF");
+
+ for (SeaTunnelRow imageRow : images) {
+ String text = (String) imageRow.getField(3);
+ Assertions.assertNotNull(text, "image element must have text");
+ Assertions.assertTrue(
+ text.startsWith("image_page_"), "image text must start
with 'image_page_'");
+
+ Integer pageNumber = (Integer) imageRow.getField(4);
+ Assertions.assertNotNull(pageNumber, "image element must have
page number");
+ Assertions.assertTrue(pageNumber >= 1, "page number must be >=
1");
+
+ // no-outline images should have null parentId
+ Assertions.assertNull(
+ imageRow.getField(6), "no-outline image must have null
parentId");
+ }
+
+ // elements should be sorted by page number
+ for (int i = 1; i < rows.size(); i++) {
+ int prevPage = (Integer) rows.get(i - 1).getField(4);
+ int currPage = (Integer) rows.get(i).getField(4);
+ Assertions.assertTrue(
+ currPage >= prevPage, "elements should be sorted by
page number");
+ }
+ } finally {
+ Files.deleteIfExists(tempPdf);
+ }
+ }
+
+ @Test
+ public void testMergeElementsIgnoresImageAndLinkWithoutParentId()
+ throws NoSuchMethodException, InvocationTargetException,
IllegalAccessException {
+ PdfReadStrategy strategy = new PdfReadStrategy();
+ Method mergeElementsMethod =
+ PdfReadStrategy.class.getDeclaredMethod(
+ "mergeElements", List.class, List.class, List.class,
List.class);
+ mergeElementsMethod.setAccessible(true);
+
+ DocumentElement heading = new DocumentElement("heading", "Section 1");
+ heading.setChildIds(new ArrayList<>());
+
+ DocumentElement image = new DocumentElement("image",
"image_page_1_pos_(100,100)");
+ image.setParentId(null);
+
+ DocumentElement link = new DocumentElement("link",
"https://example.com/orphan";);
+ link.setParentId(null);
+
+ @SuppressWarnings("unchecked")
+ List<DocumentElement> mergedElements =
+ (List<DocumentElement>)
+ mergeElementsMethod.invoke(
+ strategy,
+ Collections.singletonList(heading),
+ new ArrayList<>(),
+ Collections.singletonList(image),
+ Collections.singletonList(link));
+
+ Assertions.assertEquals(1, mergedElements.size());
+ Assertions.assertEquals("heading",
mergedElements.get(0).getElementType());
+ }
+
+ @Test
+ public void testReadDeletesTemporaryPdfCopy()
+ throws IOException, URISyntaxException, FileConnectorException {
+ URL resource =
this.getClass().getResource("/pdf_read_strategy_test.pdf");
+ String path = Paths.get(resource.toURI()).toString();
+ TrackingPdfReadStrategy pdfReadStrategy = new
TrackingPdfReadStrategy();
+ LocalConf localConf = new LocalConf(FS_DEFAULT_NAME_DEFAULT);
+ pdfReadStrategy.init(localConf);
+
+ TempCollector tempCollector = new TempCollector();
+ pdfReadStrategy.read(path, "", tempCollector);
+
+ Assertions.assertNotNull(pdfReadStrategy.getCreatedTempPdfPath());
+ Assertions.assertTrue(pdfReadStrategy.wasTempPdfDeleted());
+
Assertions.assertFalse(Files.exists(pdfReadStrategy.getCreatedTempPdfPath()));
+ }
+
+ private Path createNoOutlinePdfWithImage() throws IOException {
+ Path tempFile = Files.createTempFile("no_outline_pdf_test_", ".pdf");
+ try (PDDocument document = new PDDocument()) {
+ PDPage page = new PDPage(PDRectangle.A4);
+ document.addPage(page);
+
+ // Create a simple test image (red 50x50 square)
+ BufferedImage bufferedImage = new BufferedImage(50, 50,
BufferedImage.TYPE_INT_RGB);
+ for (int x = 0; x < 50; x++) {
+ for (int y = 0; y < 50; y++) {
+ bufferedImage.setRGB(x, y, 0xFF0000);
+ }
+ }
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ ImageIO.write(bufferedImage, "png", baos);
+ PDImageXObject pdImage =
+ PDImageXObject.createFromByteArray(document,
baos.toByteArray(), "test.png");
+
+ try (PDPageContentStream contentStream = new
PDPageContentStream(document, page)) {
+ contentStream.beginText();
+ contentStream.setFont(new
PDType1Font(Standard14Fonts.FontName.HELVETICA), 12);
+ contentStream.newLineAtOffset(50, 700);
+ contentStream.showText("This is a PDF without outline
containing an image.");
+ contentStream.endText();
+
+ contentStream.drawImage(pdImage, 50, 500, 100, 100);
+ }
+
+ // No outline is added — this PDF has no bookmarks
+ document.save(tempFile.toFile());
+ }
+ return tempFile;
+ }
+
+ private List<SeaTunnelRow> getHeadingElements(List<SeaTunnelRow> rows) {
+ return rows.stream().filter(row -> row.getField(2) !=
null).collect(Collectors.toList());
+ }
+
+ public static class LocalConf extends HadoopConf {
+ private static final String HDFS_IMPL =
"org.apache.hadoop.fs.LocalFileSystem";
+ private static final String SCHEMA = "file";
+
+ public LocalConf(String hdfsNameKey) {
+ super(hdfsNameKey);
+ }
+
+ @Override
+ public String getFsHdfsImpl() {
+ return HDFS_IMPL;
+ }
+
+ @Override
+ public String getSchema() {
+ return SCHEMA;
+ }
+ }
+
+ private static class TrackingPdfReadStrategy extends PdfReadStrategy {
+ private Path createdTempPdfPath;
+ private boolean tempPdfDeleted;
+
+ @Override
+ Path createTempPdfPath() throws IOException {
+ createdTempPdfPath = super.createTempPdfPath();
+ return createdTempPdfPath;
+ }
+
+ @Override
+ void deleteTempPdfPath(Path tempPdfPath) throws IOException {
+ super.deleteTempPdfPath(tempPdfPath);
+ tempPdfDeleted = true;
+ }
+
+ Path getCreatedTempPdfPath() {
+ return createdTempPdfPath;
+ }
+
+ boolean wasTempPdfDeleted() {
+ return tempPdfDeleted;
+ }
+ }
+}
diff --git
a/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/resources/pdf_read_strategy_test.pdf
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/resources/pdf_read_strategy_test.pdf
new file mode 100644
index 0000000000..bbc0b0d0b5
Binary files /dev/null and
b/seatunnel-connectors-v2/connector-file/connector-file-base/src/test/resources/pdf_read_strategy_test.pdf
differ
