Oracle-to-Hive Loading Without Sqoop (1) — NiFi and Load Control API Architecture

We removed Sqoop (Kylo ImportSqoop), which we used to load Oracle data into Hive, and replaced it with standard processors in Cloudera CFM 4.12 (NiFi 2.6) and a small Python service. This series covers the design and implementation in three parts. The code and the full manual are in the DataDynamics/nifi-sqoop-removal repository.

1. AS-IS: Kylo + Sqoop

The old setup worked like this.

1. Kylo ImportSqoop starts the Sqoop client and a YARN job.
2. Mappers 1 to N query the ranges split by --split-by INDEX_COLUMN from Oracle over JDBC in parallel.
3. Mapper output is stored in HDFS staging, then goes through a Hive temporary table and is written to the target with INSERT OVERWRITE.
4. Completion and failure decisions depend on Sqoop/YARN job status and logs.

Range splitting, parallel extraction and the completion decision are all bundled into a single Sqoop run. When you move this to NiFi, the first question comes up: who decides that "all Mappers have finished"?

2. What we replaced with what

We broke down what Sqoop did by role and gave each role to a different component.

NiFi is enough for moving the data. The problem is the third and fourth rows of the table. NiFi processors run independently per FlowFile, so it is hard to decide atomically, with the flow alone, whether all partitions that ran in parallel on several nodes succeeded, and to start the next step exactly once. So we moved this decision into a separate API that uses PostgreSQL transactions and CAS (conditional UPDATE).

3. TO-BE: Data Plane and Control Plane

TO-BE separates the Data Plane, which handles the actual data, from the Control Plane, which decides state and completion.

• NiFi handles only data. Oracle queries, Parquet conversion, HDFS writes, Hive SQL execution.
• The Load Control API handles only state. It decides whether all partitions are done, starts the next step only once, and records failures. The API does not move data itself.
• PostgreSQL (nifi_ops schema) is the ledger of execution state. Only the API writes to the ledger.

The Load Control API is split into two processes.

The server and worker do not talk to each other directly. They share only PostgreSQL. So you can run several of each and control contention with DB locks.

4. Processing order: PG-00 through PG-90

The NiFi canvas is laid out as follows.

NiFi root
├── PG-05 Control Receiver        # shared by all jobs
└── JOB_<JOB.KEY>                 # one per source table
    ├── PG-00 Trigger
    ├── PG-10 Run Coordinator
    ├── PG-20 Extract Worker
    ├── PG-40 Staging Validation
    ├── PG-50 Publish
    ├── PG-60 Target Validation
    ├── PG-70 Cleanup
    └── PG-90 Error and Event

The connections between PGs look like this. The key point is that the path to PG-40 does not come straight from PG-20 but goes through the API worker and PG-05. That is because the API, not NiFi, decides whether extraction is finished.

Execution nodes and parallelism

• Only Trigger (PG-00) and Cleanup (PG-70) run on the Primary Node. All other processors run on All Nodes. In a cluster, Run Once runs on every node, so if Trigger is not set to Primary Node, you get as many runs as there are nodes.
• The effective maximum parallelism of PG-20 is number of NiFi nodes × 34_Execute_Partition_Query Concurrent Tasks. This value must not exceed the Oracle connection pool and the approved number of sessions.

One set of job PGs per table

The flow is built without NiFi Registry, by the builder script in the repository (nifi-flow/deploy_job_flow.py) through the NiFi REST API. Each time you run it with a config JSON, it creates a full new PG-00 to PG-90 set under JOB_<JOB.KEY>. With 6 source tables, you get 6 job PGs with different JOB.KEY values, and only PG-05 is shared.

python3 nifi-flow/deploy_job_flow.py \
  http://<nifi-host>:<port>/nifi-api \
  /secure/path/job-insp-dtl.json

The config JSON holds per-job values: source table, extract columns, split column, business condition, partition count, Hive staging and target definitions, and so on.

"job_params": {
  "JOB.KEY": "ORACLE_INSP_DTL_DAILY",
  "SRC.OWNER": "APP",
  "SRC.TABLE": "INSP_DTL",
  "SRC.COLUMNS": "INSP_DTL_SEQ, BASE_DT, ITEM_CD, CAST(AMOUNT AS NUMBER(18,2)) AS AMOUNT, REG_TS, NOTE",
  "SRC.SPLIT.COLUMN": "INSP_DTL_SEQ",
  "SRC.BASE.WHERE": "BASE_DT = TO_DATE('${load.business.key}', 'YYYY-MM-DD')",
  "PARTITION.COUNT": "8",
  "HIVE.TARGET.DB": "dw",
  "HIVE.TARGET.TABLE": "insp_dtl",
  "TARGET.PARTITION.CLAUSE": "PARTITION (base_dt='${load.business.key}')"
}

The builder automates the repeated creation, so you don't need to copy processors in the NiFi UI. However, this is not a setup where several tables share one set of processors. If many jobs run at the same time, calculate the peak Oracle load as number of concurrent jobs × number of NiFi nodes × PG-20 Concurrent Tasks per job.

5. Keeping runs from mixing: per-run isolation

For every run, the API issues a run_id (UUID), and all output is separated by run.

<HDFS.STAGE.ROOT>/<JOB.KEY>/run_id=<run_id>/part-<partition_id>-<6-digit chunk>.parquet
<HDFS.STAGE.ROOT>/<JOB.KEY>/run_id=<run_id>/_SUCCESS
<HIVE.STAGE.DB>.<HIVE.STAGE.TABLE.PREFIX><run_id with hyphens removed>

HDFS paths and staging tables differ per run, so files from a failed run do not mix into the next run. A rerun does not fix the failed run. It runs everything again with a new run_id and a new SCN. Output from the failed run is kept as evidence for investigation during the retention period, and then PG-70 cleans it up.

6. Core guarantees

The table below summarizes what this design guarantees. Parts 2 and 3 explain how each row of this table was implemented.

7. Test results

We checked a normal run and the main failure scenarios on a 2-node Cloudera CFM 4.12 (NiFi 2.6.0) cluster with Oracle Database 23ai Free, Apache Hadoop 3.4.1, Apache Hive 4.0.1 and PostgreSQL 16. The data was 105,000 rows of APP.INSP_DTL for business date 2026-09-28, split into 8 partitions (one of them intentionally empty).

Why each scenario behaves this way is covered in Part 2 (SCN, ORA-01555) and Part 3 (duplicate runs, partial failure, PUBLISH_UNKNOWN).

Not yet verified

These are test environment results, so the following must be checked again before production.

• CDP HiveServer2 (Hive 3) and ACID managed target tables. The test used Apache Hive 4.0.1.
• Processing time with production-scale data, load on Oracle, the API and the management DB, and NiFi node failure during partition processing
• Undo retention, indexes and session limits on the production Oracle
• Cutover validation by running in parallel with AS-IS Sqoop and comparing row counts, sums and sample rows

Wrapping up

The core of replacing Sqoop with NiFi was not choosing processors. It was deciding where to put the "whole-job completion decision" that used to be bundled into a single Sqoop run. This project moved that decision out of NiFi into a small API and a PostgreSQL ledger, and as a result the NiFi flow stayed simple, doing only data processing and API calls.

In Part 2, we explain how partition queries that run at different times on several nodes read data from the same point in time, focusing on Oracle SCN and undo.