JDBC Source

JDBC Source (

) is used to read data from a relational database system (RDBMS) using JDBC (Java Database Connectivity) .

Data taken from multiple columns in an SQL query will be combined into one output field in string form using a specific separator character .

This component can be used as a replacement for Sqoop , as it is capable of reading data from relational databases and directly storing it to HDFS in flat-file or ORC format .

JDBC Source Function

Some of the main functions of JDBC Source :

Retrieving data from relational databases such as:
- MySQL
- PostgreSQL
- Oracle
- SQL Server
Supports parallel processing through partitioning.
Supports incremental data loading (fetching new data only).

Parameters of JDBC Source

1. Required Parameters

Connection parameter for JDBC connection and metadata retrieval.

Driver Class Name(*)
JDBC driver class name
Specifies the JDBC driver to use.
Driver examples:

MySQL

    com.mysql.jdbc.Driver

PostgreSQL

    org.postgresql.Driver

Oracle

    oracle.jdbc.driver.OracleDriver

Microsoft SQL Server

    com.microsoft.sqlserver.jdbc.SQLServerDriver.

URL Connection (*)
JDBC URL for database
Specifies the JDBC connection address to the database .

General format:

    jdbc:database_type://host:port/database_name

Example:

MySQL

    jdbc:mysql://localhost:3306/telecomdb

PostgreSQL

    jdbc:postgresql://localhost:5432/telecomdb

Oracle

    jdbc:oracle:thin:@localhost:1521:ORCLCDB

Microsoft SQL Server

    jdbc:sqlserver://localhost:1433;databaseName=MyDatabase;user=myUser;password=myPassword;

Username (*)
Username database
Password (*)
Password database

Query Statement(*)
SQL query to retrieve data
Fetch Size
Number of rows fetched per batch
Determines the amount of data captured per batch .
Example:
```
    Fetch Size : 1000
```
This means the system will retrieve 1000 rows of data for each query process .

Field Value Character Separator
Separator character between field values
The character used to combine multiple fields into one string . Example of character separator:

    Separator               Format
    ----------------------------------
    |                       pipe separated
    ,                       CSV    `	`
    ;                       Semicolon separated
    \t                      Tab separated
    \001                    ASCII control character
    \u0001                  Unicode control character
    char(character-code)    character-code separated

2. Partitioning Parameters

Number of Partitions
Number of parallel processes Determine the number of parallel processes .
Example:
```
    Number of Partitions : 4
```
This means that the data will be divided into 4 parallel processes .
Default value: 1
Partition & Last Updated Field
Base field used for partitioning (typically a last-updated timestamp column)
Fields used to divide data.

Usually using timestamp or ID .
Example:
```
    updated_at
```
The system will automatically calculate:
- minimum value
- maximum value
to create a partition range .

3. Incremental Processing

Incremental processing is used to retrieve only new data .

    Parameter       Description
    -----------------------------------------
    Incremental     Enable incremental queries

Example:

    Incremental : true

If active, the system only retrieves data that has never been processed before .

4. Checkpoint Environment Parameters

Checkpoints are used to store the latest information from the data retrieval process .

Driver Class Name(*)
JDBC driver class name for checkpoint access
URL(*)
JDBC URL of the checkpoint database
Table Name(*)
Checkpoint table name
Username (*)
User name for checkpoint database access
Password (*)
Password for checkpoint database access
Surrogate Key Field (*)
Field used to record the number of records that have been processed .
Default:
```
    SK_MAX
```
Last Updated Field(*)
Field to store the latest timestamp of the retrieved data .
Example:
```
    updated_at
```
Extra Param(*)
Additional parameters to uniquely identify rows of the check point table.

For example, if the checkpoint table uses a combination of field values (schema_source, database_name, and table_source) to uniquely identify rows, the extra param can be as follows:
```
    schema_source:general_information,database_name:oracle,table_source:penjualan
```

5. Dynamic Parameters

Parameters that have a flag (*)can be filled dynamically at runtime using:

    FROM_ARGUMENT(arg_name)

Example:

    URL Connection : FROM_ARGUMENT(db_url)

This allows for more flexible configuration when running the pipeline .

Replacing Sqoop with JDBC Source

JDBC Source can fully replace Sqoop by reading data from an RDBMS and loading it directly into HDFS.
Example Data Flows

Load RDBMS Data into HDFS Flat File
JDBC source→pass through Transformator→HFS Sink
→→
Load RDBMS Data into HDFS Flat File (Balanced Mode)
JDBC source→pass through Transformator→Group By→HFS Sink
→→→
Load RDBMS Data into HDFS ORC File
JDBC source→pass through Transformator→ORC Sink
→→
Load RDBMS Data into HDFS ORC File (Balanced Mode)
JDBC source→pass through Transformator→Group By→ORC Sink
→→→

Notes on Balanced Output

To generate balanced output, use:

Group By
RoundRobinPartitioning as the partition class

ORC Structure Definition via SQL Query

The structure of the ORC file is defined directly in the SQL query using column aliases and data types.
Example Query

SELECT
  rownum AS dw_sk:INT,
  kota,
  sales_id,
  item_id,
  qty AS quantity:INT,
  unit_price:DECIMAL(6,2),
  last_updated_date:TIMESTAMP
FROM penjualan;

Resulting ORC Schema

ORC Field Name	Source Column	Data Type
dw_sk	rownum	INT
kota	kota	STRING
sales_id	sales_id	STRING
item_id	item_id	STRING
quantity	qty	INT
unit_price	unit_price	DECIMAL(6,2)
last_updated_date	last_updated_date	TIMESTAMP

Supported Engines

Map Reduce	Spark 1.x RDD	Spark 2.x RDD	Spark 2.x Dataset	Apache Beam 2.29.0 Batch	Apache Beam 2.29.0 Streaming	Apache Flink 1.12.x Batch	Apache Flink 1.12.x Streaming	Apache Tez 0.9.1	PySpark