Building Snowflake Openflow Connectors: A Quick Start Connector Kit

A Snowflake Openflow Quick Start Connector Kit for Custom Database Integrations, with Complete Source Code.

TL;DR: If you cannot find a connector for your source system, you can still build a production-grade Openflow integration by building your own processors. See my example project for more details.

My example shows: Schema Discovery → Column Metadata → Data Extraction

The key is to propagate metadata through FlowFile attributes, so downstream processors can adapt automatically. Reference code (Informix example, reusable for any JDBC source) is here:

https://github.com/Unified-Honey-Inc/snowflake-openflow-connector-kit

Six months ago, I started a new chapter building my own startup and was accepted into the Snowflake Startup Accelerator.

Then I hit a problem. I needed to connect to an IBM Informix database used by a Pronto ERP system and there was no pre-built connector for Snowflake. I needed a solution to handle thousands of events per minute at peak load across 1000+ tables, with complex relationships and data types that do not map cleanly to modern systems.

I looked at the options, I could write custom Python scripts for every table? Use a third-party ETL tool that does not really support Informix? Or build it myself?

The opportunity changed how I approached building data pipelines entirely, introducing me to Snowflake Openflow and Apache NiFi.

The Openflow/NiFi solution offered visual pipelines, built-in retry and backpressure, composable processors, and a cleaner operational model than a single monolithic script.

I’ve built a custom Openflow connector for IBM Informix, then extracted the core pattern into an open-source reference project. My production implementation goes further with snapshots using watermarks, streaming CDC events for near real-time, and metadata-driven inclusion and exclusion. The example project stays focused on the reusable connector architecture so you can adapt it to almost any database.

The most annoying gotcha in my implementation was the Informix date and time handling. Some values came through as a numeric offset from 1970, so it took a bit of back and forth to get the logic right and required explicit handling in extraction and JSON serialisation so Snowflake ingested them as TIMESTAMP values.

GitHub: https://github.com/Unified-Honey-Inc/snowflake-openflow-connector-kit

Why Build Custom Openflow Connectors?

It get’s you a reliable, scalable and reusable data ingestion solution. I will show why the architecture is more elegant than you might expect.

When Standard Connectors Don’t Exist

Openflow has connectors for common databases and SaaS applications but if you are working with:

• Legacy databases (IBM Informix, Progress OpenEdge, Ingres, older On-prem systems)
• On-prem REST APIs with custom schemas
• Proprietary data formats
• Databases with specialised system catalogs
• Sources requiring non-standard authentication flows

You may need to build your own connector, although I would recommend looking at the many reusable Openflow/NiFi processors that are available.

When You Need Full Control

Even if a connector exists, custom processors are sometimes necessary for:

• Performance optimizations specific to your environment
• Source-specific type mapping decisions
• Specialised error handling and retry semantics
• Integration with internal metadata/configuration systems
• Extracting system/catalog metadata not supported out of the box

The Openflow Advantage

Why build Openflow processors instead of standalone scripts or a traditional ETL tool?

Scripts tend to become brittle, hardcoded table lists, manual schema updates, duplicated logic, and a monolithic design that is hard to evolve safely.

Openflow gives you visual flow design, built-in error handling, backpressure, connection pooling, monitoring, and operational tooling, but the real differentiator is the scalability and interoperability:

• Traditional ETL: discovery, mapping, and extraction are separate steps, metadata is not naturally reused, schema drift breaks pipelines.
• Openflow/NiFi: discovery enriches FlowFiles with metadata, mapping adds type intelligence, and extraction adapts based on context.

Simple components compose into reliable and reusable solutions.

How: Reference Project Example

The reference project demonstrates an architecture that works and can be adapted across different databases:

[Schema Discovery]  →   [Column Metadata]   →  [Data Extraction]
        ↓                       ↓                      ↓
   "What exists?"       "How do types map?"   "Extract and stream"

Three processors, each with a single responsibility:

1. Schema Discovery: queries the system catalog to find tables (and optionally views)
2. Column Metadata: extracts column details and maps types to Snowflake equivalents
3. Data Extraction: executes SQL and streams results as NDJSON

Separating logic into components, you can run discovery alone for inventory and add metadata extraction for type mapping. Chain all three for migration or ingestion using the same processors, composed differently to solve different problems.

What Production Adds (Beyond the Reference Repo)

In my production flows I add a couple of steps to make it fully metadata-driven:

1) Generate strongly typed Snowflake destination tables

[Schema Discovery] → [Column Metadata] → [Create/Alter Snowflake Tables]

2) Metadata-driven extraction with state / watermarks

[CDC/Watermark State] → [Openflow Processor] → [Incremental Extraction] → [Snowpipe/Snowpipe Streaming]

The reference repo stays focused on the reusable building blocks, so you can adapt it without dragging in environment-specific decisions.

Why Metadata Flow Matters

Context flows through FlowFile attributes, the discovery processor doesn’t just output a table list. It tells the next processor “I found 300 tables in the ‘ERP’ database”. The metadata processor adds “Here’s how SERIAL maps to NUMBER(38,0)”. The extraction processor receives both pieces of context and generates optimized SQL.

Each processor is purpose-built for a task, as a system, they map your data architecture automatically, adapting to your workflows in real-time.

What: Things to watch

Building Openflow/NiFi connectors requires solving a few issues that are not obvious until you hit them. Here are the big three.

#1: The Classloader Mystery

A common early failure looks like this:

java.lang.ClassCastException: 
org.apache.nifi.dbcp.DBCPService cannot be cast to DBCPConnectionPool

This happens because Openflow/NiFi runs processors in isolated classloaders. A Controller Service interface can be loaded in a different classloader than your processor, so even “identical” types are not compatible at runtime.

DBCPService (DBCPConnectionPool) provides a background service that manages the JDBC connections to source databases, it provides an ondemand connection pool for processors needing a database connection. An elegant design that keeps processors focused on their task and separates repeatable logic.

The solution is simple when using reflection:

public class DBCPServiceHelper {
    public static Connection getConnection(ProcessContext context,
                                           PropertyDescriptor property) {
        final ControllerService service = context.getProperty(property)
                                                 .asControllerService();
        try {
            if (service instanceof DBCPService) {
                return ((DBCPService) service).getConnection();
            }

            // Handle classloader isolation via reflection
            Method getConnectionMethod = service.getClass()
                                               .getMethod("getConnection");
            return (Connection) getConnectionMethod.invoke(service);
        } catch (Exception e) {
            throw new ProcessException("Failed to get connection", e);
        }
    }
}

This utility works for any JDBC-based connector.

#2: Metadata Propagation

Your discovery processor finds tables, but how do downstream processors know what was discovered?

FlowFile Attributes

Every piece of context needs to flow through attributes, your data flows between processors in FlowFiles. Additional metadata can be added to a FlowFile to provide instructions and context to the next processor. An example is the destination table name in Snowflake for the data:

Map<String, String> attributes = new HashMap<>();
attributes.put("db.name", databaseName);
attributes.put("table.count", String.valueOf(tables.size()));
attributes.put("sf.table", "TABLE_LIST");

flowFile = session.putAllAttributes(flowFile, attributes);

Without this, processors cannot compose, each operates in isolation, with attributes on FlowFiles the pipeline becomes adaptive and avoids hardcoding.

#3: Type System Mapping

Every database has unique types, Informix SERIAL is not the same as Oracle NUMBER or PostgreSQL BIGSERIAL. You need explicit mappings:

private String mapToSnowflake(String sourceType, int precision, int scale, int length) {
    switch (sourceType.toUpperCase()) {
        case "SERIAL":
            return "NUMBER(38,0)";
        case "DECIMAL":
            return String.format("NUMBER(%d,%d)", precision, scale);
        case "DATETIME":
            return "TIMESTAMP_NTZ";
        case "VARCHAR":
            return String.format("VARCHAR(%d)", length);
        case "JSON":
            return "VARIANT";
        default:
            return "VARCHAR(16777216)";
    }
}

The principle is conservative mapping and avoid data loss, use VARIANT for semi-structured types, and default unknowns to a wide VARCHAR.

These decisions matter, get them wrong and you lose data or fail validation, if you get them right once, every connector you build benefits from that knowledge.

Implementation: Schema Discovery Processor

The first processor discovers what tables exist (example, IfxGetTableList.java).

onTrigger Function

Openflow/NiFi will call the onTrigger function when there is work to do, this can be based on:

Event-Driven: It runs as soon as a FlowFile enters the processor’s input queue. Timer-Driven: It runs at a specific interval (e.g., every 10 seconds), which is useful for “source” processors that fetch data from external APIs like Salesforce or HubSpot. CRON-Driven: It can be scheduled to run at specific times of the day (e.g., 2:00 AM every Tuesday).

A Timer-Driven approach can use 0 Seconds to effectively run as fast as possible, immediately after finishing its previous task. It can be made to automatically wait 10ms before trying again or set to 10 minutes or 3 hours depending on your requirements.

Schema Discovery Example

@Override
public void onTrigger(ProcessContext context, ProcessSession session) {

    // Source-style processor: create a new FlowFile to hold the discovered table list payload.
    FlowFile flowFile = session.create();

    // Build the database-specific discovery SQL (system catalog query) using any configured filters.
    String sql = buildTableDiscoverySQL(tableFilter);

    // Get a JDBC connection from the configured DBCP service. Helper handles classloader isolation.
    // try-with-resources ensures the connection is returned/closed properly.
    try (Connection conn = DBCPServiceHelper.getConnection(context, DBCP_SERVICE)) {

        // Execute the discovery query and materialise results as a list of row maps.
        // Consider streaming directly to JSON if this list can become large.
        List<Map<String, Object>> tables = queryTables(conn, sql);

        // Convert the discovered tables to a JSON array so downstream processors can parse content easily.
        // Each entry is one table record from the system catalog query.
        JSONArray jsonArray = new JSONArray();
        for (Map<String, Object> table : tables) {
            JSONObject json = new JSONObject(table);

            // Add an ingestion timestamp for traceability and auditing.
            // Use a consistent field name across processors if you rely on it downstream.
            json.put("INGESTED_AT", Instant.now().toString());

            jsonArray.put(json);
        }

        // Write the JSON array as the FlowFile content (UTF-8).
        // Consider setting mime.type=application/json and a filename attribute for observability.
        flowFile = session.write(flowFile, out ->
            out.write(jsonArray.toString().getBytes(StandardCharsets.UTF_8)));

        // Attach lightweight context as FlowFile attributes for downstream processors:
        // - db.name: which source database/schema this list came from
        // - table.count: how many objects were discovered
        // - sf.table: optional routing hint for target table naming in Snowflake
        Map<String, String> attributes = new HashMap<>();
        attributes.put("db.name", databaseName);
        attributes.put("table.count", String.valueOf(tables.size()));
        attributes.put("sf.table", "TABLE_LIST");

        flowFile = session.putAllAttributes(flowFile, attributes);

        // Route the FlowFile to success for downstream processing (split by table, extract columns, etc).
        // In production, add a catch block to route to REL_FAILURE and remove the created FlowFile on error.
        session.transfer(flowFile, REL_SUCCESS);
    }
}

Key points:

• Uses DBCPServiceHelper to avoid classloader pitfalls
• Outputs a JSON array of tables
• Sets FlowFile attributes for downstream processors
• Adds an INGESTED_AT timestamp for traceability

What Changes Per Database

The SQL query is database-specific. Everything else stays the same.

Informix (tables only):

SELECT tabname, tabid, tabtype
FROM systables
WHERE tabid > 99
  AND tabtype = 'T'
ORDER BY tabname;

Oracle (schema-owned tables):

SELECT owner, table_name
FROM all_tables
WHERE owner = UPPER('YOUR_SCHEMA_NAME')
ORDER BY table_name;

PostgreSQL (exclude system schemas):

SELECT schemaname, tablename
FROM pg_tables
WHERE schemaname NOT IN ('pg_catalog','information_schema')
ORDER BY schemaname, tablename;

The processor logic stays mostly identical, only the system catalog query changes.

Implementation: Column Metadata Processors

The second processor (for example, IfxGetColumnsList.java) receives a table list and extracts column details for each table. It does not need to know tables in advance, it processes whichever tables the incoming FlowFile describes.

Column Metadata Example

@Override
public void onTrigger(ProcessContext context, ProcessSession session) {
    // Pull the next FlowFile from the incoming queue. If none, return quickly.
    FlowFile flowFile = session.get();
    if (flowFile == null) return;

    // The upstream processor should set table.name. Consider validating this and routing to failure if missing.
    String tableName = flowFile.getAttribute("table.name");

    // Build a database-specific column metadata query for the current table.
    String sql = buildColumnQuerySQL(tableName);

    // Obtain a JDBC connection via the DBCP service helper (handles classloader isolation cases).
    // try-with-resources ensures the Connection is returned/closed properly.
    try (Connection conn = DBCPServiceHelper.getConnection(context, DBCP_SERVICE)) {

        // Execute the query and iterate the result set to build a JSON array of column metadata.
        // Note: consider closing rs (and statement) explicitly, or ensure executeQuery does that internally.
        ResultSet rs = executeQuery(conn, sql);

        JSONArray columns = new JSONArray();
        while (rs.next()) {
            // Read source column metadata fields as provided by the system catalog query.
            String colName = rs.getString("column_name");
            String sourceType = rs.getString("data_type");
            int precision = rs.getInt("precision");
            int scale = rs.getInt("scale");

            // Construct a canonical JSON record per column, normalising names to uppercase for Snowflake conventions.
            // Consider handling null colName/sourceType defensively.
            JSONObject column = new JSONObject();
            column.put("COLUMN_NAME", colName.toUpperCase());
            column.put("SOURCE_TYPE", sourceType);

            // Map the source type to an explicit Snowflake type (centralised mapping logic).
            // Ensure mapToSnowflake uses the right sizing field (length vs scale) for string types.
            column.put("SNOWFLAKE_TYPE", mapToSnowflake(sourceType, precision, scale));

            // Preserve raw source metadata as separate fields for auditing / troubleshooting.
            column.put("PRECISION", precision);
            column.put("SCALE", scale);

            columns.put(column);
        }

        // Replace the FlowFile content with the JSON array payload (UTF-8 encoded).
        // Consider setting an attribute like mime.type=application/json and row/column counts for downstream use.
        flowFile = session.write(flowFile, out ->
            out.write(columns.toString().getBytes(StandardCharsets.UTF_8)));

        // Route to success. In production, wrap in try/catch and route to REL_FAILURE on errors,
        // so FlowFiles are not silently dropped and the flow can retry or quarantine.
        session.transfer(flowFile, REL_SUCCESS);
    }
}

The mapToSnowflake method contains all the database-specific knowledge about how types translate, this is where you encode your understanding of both systems.

Implementation: Data Extraction Processor

The third processor (IfxExecuteSQL.java) pulls the actual data. Executes custom SQL statements against Informix and outputs results optimized for Snowflake ingestion.

Use cases include initial loads, incremental extraction (watermark SQL), ad-hoc queries, and data migration from Informix to Snowflake.

Execute SQL Query Results Example

while (rs.next() && rowCount < maxRows) {

    JSONObject dataObject = new JSONObject();

    for (int i = 1; i <= columnCount; i++) {
        String columnName = metaData.getColumnName(i).toUpperCase();
        dataObject.put(columnName, rs.getObject(i));
    }
    // Wrap in record with metadata
    JSONObject record = new JSONObject();

    record.put("DATA", dataObject);
    record.put("TABLE_NAME", extractTableName(sql));
    record.put("INGESTED_AT", Instant.now().toString());

    ndjsonBatch.append(record.toString()).append("\n");

    if (++rowCount % batchSize == 0) {
        createBatchFlowFile(session, ndjsonBatch.toString());
        ndjsonBatch.setLength(0);
    }
}

Key points:

• Streams results (doesn’t load entire dataset into memory)
• Batches into configurable sizes (default 1000 rows)
• Uses NDJSON format with DATA field
• Sets proper attributes on each batch FlowFile

Why batching matters: you do not want to load millions of rows into memory. Batch FlowFiles let Openflow ingest incrementally and recover cleanly. The processor creates a new FlowFile every N rows, allowing incremental processing.

How to Adapt for Your Database

To build a connector for a different database, here is what changes and what stays the same.

Copy directly:

• DBCPServiceHelper (works for any JDBC source)
• The example processors as a template
• Attribute propagation pattern
• NDJSON output format
• Batch processing logic

Change these:

• System catalog SQL queries
• Type mappings in mapToSnowflake
• JDBC URL format / driver details
• (Optional) add a processor to read watermarks and inclusion/exclusion rules from Snowflake.

Example: Adapting for other systems

Change the system catalog query:

// Informix
SELECT tabname FROM systables WHERE tabid > 99

// PostgreSQL
SELECT tablename FROM pg_tables WHERE schemaname = 'public'

// Oracle
SELECT table_name FROM all_tables WHERE owner = 'YOUR_SCHEMA_NAME'

// SQL Server
SELECT name AS tablename FROM sys.tables WHERE is_ms_shipped = 0

Change the type mappings:

case "BIGSERIAL":
    return "NUMBER(19,0)";
case "UUID":
    return "VARCHAR(36)";
case "JSONB":
    return "VARIANT";

What took me a while to learn should take you a few days to adapt once you have the pattern.

What You Get

The Informix reference implementation includes:

Complete source code

• Three processors (discovery, metadata, extraction)
• DBCPServiceHelper
• Example flow definitions
• Maven build configuration

Documentation

• Getting started guide
• README with examples
• Snowflake initialisation scripts
• JSON output samples

Real-world usage:

• Handles 300+ table ERPs
• 25+ type mappings
• Incremental load patterns
• Error handling

This isn’t just sample code, the code will extract data from actual ERP systems.

The Bigger Picture

Once you see this pattern, you start recognising it everywhere:

• REST API endpoint discovery → schema extraction → data retrieval
• File system directory listing → file metadata → content extraction
• Message queue topic discovery → schema registry → consumption

It is about composable components that pass context forward. That is what makes custom connector development tractable.

Getting Started

Clone the repository:

git clone https://github.com/Unified-Honey-Inc/snowflake-openflow-connector-kit.git

Build the connector:

cd "Openflow Connector for IBM Informix"
mvn clean install

The result is a NAR file you upload to Openflow. Import the example flow definition, configure your database connection, and run.

Full documentation: Getting Started Guide on GitHub

Contributing

If you build a connector using this pattern, share what you learned and contribute improvements back (edge cases, type mappings, catalog queries, better defaults), additional processor examples would be great.

For questions, open a GitHub discussion. For bugs, open an issue. For improvements, submit a pull request.

Simon Yeoman is founder of [Unified Honey Inc](https://unifiedhoney.com?medium=Building Snowflake Openflow Connectors: A Quick Start Connector Kit)., building and automating data solutions. Connect on LinkedIn or GitHub.

Tags: #Snowflake #Openflow #ApacheNiFi #DataEngineering #OpenSource #DatabaseConnector #StarterKit #Informix