Skip to main content
Last updated:|... min read
Bug iconOpen OLake issuesBug iconOpen OLake doc issue

Overview

The OLake Go S3 Source connector ingests data from Amazon S3 or S3-compatible storage (MinIO, LocalStack). It offers features like parallel chunking, checkpointing, and automatic resume for failed full loads. This connector can be used within the OLake UI or run locally via Docker for open-source workflows.

Key Features​

  • Multi-Format Support: CSV files (.csv, .csv.gz), JSON files in JSONL and Array formats (.json, .jsonl, .json.gz), XML files (.xml, .xml.gz) and Parquet files

  • Automatic Compression Handling: Transparent gzip decompression (.gz files)

  • Folder-Based Streams: Groups files by top-level folder into logical streams. Files are automatically grouped into streams based on folder structure. An example of bucket structure is provided below:

    s3://my-bucket/data/
    ├── users/
    │   ├── 2024-01-01/users.parquet
    │   └── 2024-01-02/users.parquet
    ├── orders/
    │   ├── 2024-01-01/orders.parquet
    │   └── 2024-01-02/orders.parquet
    └── products/
        └── products.csv.gz

    In the above example, as a result we get 3 streams created - users, orders, products.

  • S3-Compatible Services: AWS S3, MinIO, LocalStack, and other S3 APIs

Sync Modes Supported​

  • Full Refresh
  • Full Refresh + Incremental
How Incremental Sync Works

The S3 connector uses S3's LastModified timestamp as a cursor for incremental syncs:

How it works:

  1. Discovery phase adds _last_modified_time field to each stream
  2. During sync, each record is injected with the file's LastModified timestamp
  3. State file tracks the latest _last_modified_time per stream
  4. Subsequent syncs only process files with LastModified > last_synced_timestamp

State Example:

{
  "users": {
    "_last_modified_time": "2024-01-15T10:30:00Z"
  },
  "orders": {
    "_last_modified_time": "2024-01-15T11:45:00Z"
  }
}
File Modifications

If a file's content changes and is re-uploaded to S3, it will be re-synced in incremental mode because S3 updates the LastModified timestamp.

Prerequisites​

Version Prerequisites​

  • AWS S3 (any version) or S3-compatible service (MinIO 2020+, LocalStack 0.12+)

Connection Prerequisites​

  • Read access to S3 bucket (s3:ListBucket, s3:GetObject)

  • AWS Authentication: IAM roles, environment variables, instance profiles (recommended), or static access key and secret key credentials

  • Network connectivity to S3 endpoint

  • IAM policy for S3 source access:

    s3-source-iam-policy.json
    {
      "Version": "2012-10-17",
      "Statement": [
        {
          "Effect": "Allow",
          "Action": [
            "s3:ListBucket"
          ],
          "Resource": [
            "arn:aws:s3:::<YOUR_S3_BUCKET>"
          ]
        },
        {
          "Effect": "Allow",
          "Action": [
            "s3:GetObject"
          ],
          "Resource": [
            "arn:aws:s3:::<YOUR_S3_BUCKET>/*"
          ]
        }
      ]
    }
    • Replace <YOUR_S3_BUCKET> with your actual S3 bucket name.
    • This policy provides read-only access required for the OLake Go S3 source connector.

After initial Prerequisites are fulfilled, the configurations for S3 can be configured.

XML File Handling​

The following behavior applies when XML is selected as the file format.

1. Normalization OFF​

When normalization is set to OFF in the Streams section, Row Identifier is ignored. OLake Go parses the XML into a JSON tree and stores it as a JSON string in the data column. Nested elements are stored as stringified JSON within that structure.

note

If the Row Identifier entered in the source configuration is not present in the XML file, stream discovery will fail.

Example input:

<orders>
  <order id="1001" status="NEW">
    <order_date>2024-08-01</order_date>
    <customer>
      <id>CUST_55</id>
      <name>Alice</name>
    </customer>
  </order>
</orders>

Output:

{"_last_modified_time":"2026-08-19T11:17:30Z","orders":{"order":"{\"_id\":\"1001\",\"_status\":\"NEW\",\"customer\":\"{\\\"id\\\":\\\"CUST_55\\\",\\\"name\\\":\\\"Alice\\\"}\",\"order_date\":\"2024-08-01\"}"}}

2. Normalization ON​

When normalization is set to ON in the Streams section, OLake Go performs level 0 flattening and stores the data as a table. You must also set Row Identifier in the source configuration to the tag that encloses each record.

note

If the Row Identifier is invalid (not present in the XML file), stream discovery will fail.

Example input (with row_identifier = order):

<orders>
  <order id="1001" status="NEW">
    <order_date>2024-08-01</order_date>
    <customer>
      <id>CUST_55</id>
      <name>Alice</name>
    </customer>
  </order>
</orders>

Output table:

_id_statusorder_datecustomer
1001NEW2024-08-01{"id":"CUST_55","name":"Alice"}

3. Level 0 Flattening Rules​

When normalization is ON, OLake Go applies level 0 flattening only. Using the example above:

  • Attributes on the row identifier element become columns prefixed with _. For <order id="1001" status="NEW">, id and status become columns _id and _status.
  • Direct child elements become columns using their tag name as-is. For example, order_date becomes column order_date.
  • Nested child elements become columns using their tag name, but their contents are not flattened further. The value is stored as a JSON string. For example, customer becomes column customer with value {"id":"CUST_55","name":"Alice"}.

4. Repeated Elements and Row Identifier​

Choosing the correct Row Identifier matters when XML contains repeated elements.

Example input:

<items>
  <item>Apple</item>
  <item>Banana</item>
</items>
Row IdentifierResult
itemTable with column item and two rows: Apple, Banana
itemsTable with column item and one row: Banana (Apple gets overwritten)
Row Identifier Selection

Always set Row Identifier to the tag that encloses each individual record, not the parent wrapper element. An incorrect value can cause data loss from overwrites.

5. Namespaced Elements​

When XML elements use namespaces, column names include the namespace prefix.

Example input (with normalization ON):

<h:order id="1">
  <h:price>10</h:price>
  <f:price xmlns:f="http://f.ns">20</f:price>
</h:order>

Output table columns: _id, h_price, f_price

The namespace prefix is included in the column name to avoid collisions between elements that share the same local name (for example, h:price and f:price).

Configuration​

1. Navigate to the Source Configuration Page​

  1. Complete the OLake UI Setup Guide
  2. After logging in to the OLake UI, select the Sources tab from the left sidebar
  3. Click Create Source on the top right corner
  4. Select S3 from the connector dropdown
  5. Provide a name for this source

2. Provide Configuration Details​

  • Enter S3 credentials.
    Form for creating a S3 source in OLake, showing fields for endpoint configuration, authentication, and connection options
FieldDescriptionExample Value
Bucket Name requiredS3 bucket name where Iceberg table data and metadata files will be stored.my-data-warehouse
Region requiredAWS Region where your S3 bucket exists. Required so OLake Go calls the correct regional S3 endpoint.us-east-1
Path PrefixOptional path prefix to filter filesdata/
Access Key IDAWS access key ID for authentication. Optional if using IAM roles or instance profiles.<YOUR_KEY>
Secret Access KeyAWS secret access key for authentication. Optional if using IAM roles or instance profiles.<YOUR_SECRET>
File Format requiredSelect the format of files in the bucket to sync. OLake Go uses this to determine how to read and process files during sync.
  • Parquet
  • XML
  • JSON
  • CSV
Compression TypeCompression format of the files.
  • none
  • gzip
  • zip
File PatternRegular expression to filter files by name..*\.csv$
Custom EndpointS3 API endpoint for writing Iceberg data files. Optional for AWS S3 but required for S3-compatible storage like MinIO.https://glue.ap-south-1.amazonaws.com
Max ThreadsMaximum number of worker threads the connector can spin up for parallel tasks.10
Retry CountNumber of retry attempts for failures.3
AWS Authentication
  • Access Key ID and Secret Access Key are optional for AWS S3. If omitted, the driver uses AWS default credential chain (IAM roles, environment variables, instance profiles, ECS task roles). This is the recommended approach for production deployments. If you provide one credential, you must provide both.
  • Access Key ID, Secret Access Key and Endpoint is required for non AWS S3 services.

3. Format-Specific Configuration​

Additional fields for CSV files:

FieldDescriptionDefault
DelimiterField separator character,
Has HeaderWhether CSV files contain a header rowtrue
Skip RowsNumber of rows to skip at the beginning of CSV files0
Quote CharacterQuote character for CSV files"

Compression: Automatically detected from file extension (.csv.gz = gzipped)

4. Test Connection​

  • Once the connection is validated, the S3 source is created. Jobs can then be configured using this source
  • In case of connection failure, refer to the Troubleshooting section

Data Type Mapping​

File FormatSource Data TypeDestination Data TypeNotes
CSVInferred from datastringAll CSV fields initially treated as strings
CSVNumeric patternsint, bigint, doubleInteger and floating-point numbers auto-detected
CSVISO 8601 datestimestamptzDate/datetime strings converted to timestamp
CSVBoolean valuesbooleantrue/false strings converted to boolean
JSONstringstringJSON string fields
JSONnumber (integer)bigintJSON integer values
JSONnumber (float)doubleJSON floating-point values
JSONbooleanbooleanJSON boolean values
JSONobject, arraystringNested objects/arrays serialized to JSON strings
JSONnullstringNull values converted to empty strings
ParquetSTRING, BINARYstringParquet string types
ParquetINT32, INT64int, bigintParquet integer types
ParquetFLOAT, DOUBLEfloat, doubleParquet floating-point types
ParquetBOOLEANbooleanParquet boolean type
ParquetTIMESTAMP_MILLIStimestamptzParquet timestamp types
ParquetDATEdateParquet date type
ParquetDECIMALfloatParquet decimal types converted to float64
XMLtext content, attributesstringAll scalar element text and attribute values treated as strings
XMLnested elementsstringNested elements serialized to JSON strings
XMLempty or nullstringEmpty or null values treated as nullable string
All Formats_last_modified_timetimestamptzS3 LastModified metadata (added by connector)
Schema Inference
  • CSV: Uses AND logic - examines all sampled rows to determine most restrictive type
  • JSON: Auto-detects types from JSON primitives
  • Parquet: Schema read directly from file metadata (no inference needed)
  • XML: All values treated as strings; no datatype inference. Nested elements stored as JSON strings.
timestamptz timezone

OLake Go always ingests timestamp data in UTC format, independent of the source timezone.

Date and Time Handling​

During transfer, values in date, time, and timestamp columns are modified to ensure valid calendar ranges and destination compatibility.

  • Case I (Year 0000):
    Source dates with year 0000 are not valid in most destinations, so we change them to the epoch start date.
    Example: 0000-05-10 → 1970-01-01
  • Case II (Year > 9999):
    Extremely large years are capped at 9999. The month and date are not affected.
    Examples: 10000-03-12 → 9999-03-12
  • Case III (Invalid month/day):
    When the month or day exceeds valid ranges (i.e. month > 12 or day > 31), or the combined date is invalid, the value is replaced with the epoch start date.
    Examples: 2024-13-15 → 1970-01-01, 2023-04-31 → 1970-01-01
Note

These rules apply to date, time, and timestamp columns during transfer.

Troubleshooting​

1. Connection Failed - Access Denied​

ERROR failed to list objects: AccessDenied: Access Denied

Cause: Insufficient IAM permissions or incorrect credentials

Solution:

  • If using static credentials: Verify access key and secret key are correct
  • If using IAM roles: Ensure the IAM role has proper S3 permissions attached
  • Check IAM policy includes: 
    {
      "Effect": "Allow",
      "Action": ["s3:ListBucket", "s3:GetObject"],
      "Resource": [
        "arn:aws:s3:::bucket-name",
        "arn:aws:s3:::bucket-name/*"
      ]
    }
  • AWS Credential Chain Order:
    1. Static credentials in config (access_key_id, secret_access_key)
    2. Environment variables (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY)
    3. IAM role attached to EC2 instance/ECS task
    4. AWS credentials file (~/.aws/credentials)
  • For MinIO/LocalStack: Ensure credentials match server configuration

2. No Streams Discovered​

Cause: Files not organized in folders or incorrect path_prefix

Solution:

  • S3 connector requires folder structure: bucket/prefix/stream_name/files
  • Check path_prefix matches your structure
  • Verify file extensions match format (.csv, .json, .parquet)
  • Example command to verify: 
    aws s3 ls s3://bucket-name/prefix/ --recursive

3. Schema Inference Failed - CSV​

ERROR failed to infer schema: invalid delimiter or header configuration

Cause: Incorrect CSV configuration

Solution:

  • Verify has_header matches file (check first row)
  • Check delimiter is correct (, vs ; vs \t)
  • Ensure all rows have same column count
  • Test with a small sample file first

4. JSON Format Not Detected​

Cause: Mixed JSON formats in same folder or invalid JSON

Solution:

  • Keep JSONL and JSON Array formats in separate folders/streams
  • Validate JSON syntax: jq . < file.json
  • Ensure consistent field names across records
  • Check for trailing commas or syntax errors

5. Parquet File Cannot Be Read​

ERROR failed to read parquet schema: not a parquet file

Cause: Corrupted file or invalid Parquet format

Solution:

  • Verify file with parquet-tools: parquet-tools schema file.parquet
  • Check file wasn't corrupted during upload
  • Ensure file extension is .parquet (not .pq or other)
  • Re-upload file from source

6. Incremental Sync Not Working​

Cause: State file not persisted or incorrect sync_mode

Solution:

  • Verify state.json file location is writable
  • Check catalog has sync_mode: "incremental"
  • Ensure cursor_field: "_last_modified_time" is set
  • Confirm state file is being passed to sync command: 
    --state /path/to/state.json

7. MinIO Connection Timeout​

ERROR dial tcp: i/o timeout

Cause: Network connectivity or incorrect endpoint

Solution:

  • Check MinIO is running: docker ps | grep minio
  • Test endpoint: curl http://localhost:9000/minio/health/live
  • Verify endpoint format: http://hostname:9000 (include protocol)
  • For Docker: Use container name instead of localhost

8. Files Not Syncing Despite Being Present​

Cause: File extension mismatch or compression not detected

Solution:

  • Ensure file extensions match format:
    • CSV: .csv or .csv.gz
    • JSON: .json, .jsonl, .json.gz, .jsonl.gz
    • Parquet: .parquet
  • Check file size is non-zero: aws s3 ls s3://bucket/prefix/ --recursive --human-readable
  • Verify files are in correct folder structure

9. Out of Memory Errors​

FATAL runtime: out of memory

Cause: Too many large files processed concurrently

Solution:

  • Reduce max_threads in configuration (try 3-5)
  • Process fewer streams at once
  • Split very large files (>5GB) before upload
  • Increase container memory limits

10. Permission Denied - LocalStack​

Cause: LocalStack IAM policy simulator

Solution:

  • LocalStack accepts any credentials by default (test/test)
  • Ensure endpoint is correct: http://localhost:4566
  • Check LocalStack is running: docker ps | grep localstack
  • Verify bucket exists: awslocal s3 ls

If the issue is not listed here, post the query on Slack to get it resolved within a few hours.

Changelog​

Date of ReleaseVersionDescription
TBDv0.4.0Initial S3 source connector release


💡 Join the OLake Community!

Got questions, ideas, or just want to connect with other data engineers?
👉 Join our Slack Community to get real-time support, share feedback, and shape the future of OLake together. 🚀

Your success with OLake is our priority. Don’t hesitate to contact us if you need any help or further clarification!

Join our growing community

GitHub
Slack
Twitter
LinkedIn
YouTube