Enhancing Apache NiFi 2.x with Python Processors - 12Sept2025

https://github.com/tspannhw/conferenc...

Community over Code 2025 US

https://communityovercode.org/schedule/

Timothy Spann
Enhancing Apache NiFi 2.x with Python Processors
In this talk, I will delve into the world of Apache NiFi 2.0 Python processors, exploring the capabilities they offer and demonstrating how to build custom processors to enhance your data processing pipelines.

By the end of this talk, participants will have a comprehensive understanding of building and optimizing Apache NiFi 2.0 Python processors, enabling them to integrate Python seamlessly into their data processing workflows.

This session is suitable for data engineers, architects, and anyone interested in harnessing the combined power of Apache NiFi and Python for efficient data integration and flow management. One of the main uses is to build prompts and call open LLM and AI. NiFi excels at integration, I will cover some interesting sources, sinks and enrichments and show when Python is helpful.

Lake Nokomis Fri 10:20 am - 11:00 am

https://github.com/tspannhw/Example-R...
https://github.com/tspannhw/FLaNK-pyt...
https://github.com/tspannhw/CortexAIS...
https://github.com/tspannhw/conferenc...
https://www.slideshare.net/slideshow/...

Enhancing Apache NiFi 2.x with Python Processors
For the more technical audience, this talk showcased how to extend Apache NiFi's functionality using custom Python processors. It's a great way to integrate specialized logic and libraries directly into your dataflows.

Travel Advisories (From RSS)
ICEBERG TABLE TRAVELADVISORIES

Local Transit (From GTFS-RT protobuf or SIRI JSON)
ICEBERG TABLE VEHICLEPOSITIONS
ICEBERG TABLE SUBWAY
ICEBERG TABLE SERVICEALERTS
ICEBERG TABLE ICYMTA
ICEBERG TABLE MTABUSVEHICLEMONITORING ← TODO

Flights / Airport / ADS-B / Planes (From JSON and Antenna)
ICEBERG TABLE FLIGHT_DATA_ICEBERG
ICEBERG TABLE PLANES

Weather (From XML)
ICEBERG TABLE NOAAWEATHER

Air Quality (From JSON)
ICEBERG TABLE AQ
ICEBERG TABLE AQFORECAST

Local Sensors (From RPI as JSON)
ICEBERG TABLE SENSORS

Social Media BlueSky (From JSON) ←- TODO HASHTAG?
ICEBERG TABLE SOCIALMEDIA

CREATE OR REPLACE ICEBERG TABLE traveladvisories (
title VARCHAR,
link VARCHAR,
description VARCHAR,
published VARCHAR,
author VARCHAR,
category VARCHAR
)
EXTERNAL_VOLUME = 'TRANSCOM_TSPANNICEBERG_EXTVOL'
CATALOG = 'SNOWFLAKE'
BASE_LOCATION = 'traveladvisories/';
PARTITION BY (category);

create or replace ICEBERG TABLE DEMO.DEMO.FLIGHT_DATA_ICEBERG (
RCTIMESTAMP STRING,
UUID STRING,
CREATEDDATE STRING,
ICAO24 STRING,
CALLSIGN STRING,
FIRSTSEEN STRING,
LASTSEEN STRING,
ESTDEPARTUREAIRPORT STRING,
ESTARRIVALAIRPORT STRING,
ESTDEPARTUREAIRPORTHORIZDISTANCE STRING,
ESTARRIVALAIRPORTHORIZDISTANCE STRING,
ESTDEPARTUREAIRPORTVERTDISTANCE STRING,
ESTARRIVALAIRPORTVERTDISTANCE STRING,
ARRIVALAIRPORTCANDIDATESCOUNT STRING,
DEPARTUREAIRPORTCANDIDATESCOUNT STRING,
AIRPORT STRING,
DATAURL STRING,
PLANEURL STRING,
PLAINDETAILSURL STRING,
AIRCRAFTMODE STRING,
AIRCRAFTTYPE STRING,
AIRCRAFTMANUFACTURER STRING,
AIRCRAFTURLPHOTO STRING,
AIRCRAFTURLPHOTOTHUMB STRING,
AIRCRAFTOWNERCOUNTRY STRING,
AIRCRAFTREGISTEREDOWNER STRING,
AIRCRAFTICAOTYPE STRING,
AIRCRAFTREGISTRATION STRING
)COMMENT='The table contains records of flight data, specifically aircraft movements, tracked by unique identifiers. Each record includes details about the flight''s estimated departure and arrival airports, as well as estimated distances and airport candidates. The records cover various aspects of flight operations, including estimated arrival and departure times.'

EXTERNAL_VOLUME = 'TRANSCOM_TSPANNICEBERG_EXTVOL'
CATALOG = 'SNOWFLAKE'
BASE_LOCATION = 'flightdata/';

create or replace ICEBERG TABLE DEMO.DEMO.NOAAWEATHER(
CREDIT STRING,
CREDIT_URL STRING,
IMAGE_URL STRING,
IMAGE_TITLE STRING,
IMAGE_LINK STRING,
STATION_ID STRING,
LOCATION STRING,
LATITUDE FLOAT,
LONGITUDE FLOAT,
SUGGESTED_PICKUP STRING,
SUGGESTED_PICKUP_PERIOD FLOAT,
OBSERVATION_TIME STRING,
OBSERVATION_TIME_RFC822 STRING,
WEATHER STRING,
TEMPERATURE_STRING STRING,
TEMP_F FLOAT,
TEMP_C FLOAT,
RELATIVE_HUMIDITY FLOAT,
WIND_STRING STRING,
WIND_DIR STRING,
WIND_DEGREES FLOAT,
WIND_MPH FLOAT,
WIND_KT FLOAT,
PRESSURE_IN FLOAT,
DEWPOINT_STRING STRING,
DEWPOINT_F FLOAT,
DEWPOINT_C FLOAT,
VISIBILITY_MI FLOAT,
ICON_URL_BASE STRING,
TWO_DAY_HISTORY_URL STRING,
ICON_URL_NAME STRING,
OB_URL STRING,
DISCLAIMER_URL STRING,
COPYRIGHT_URL STRING,
PRIVACY_POLICY_URL STRING,
CREATED_AT TIMESTAMP_NTZ,
UPDATED_AT TIMESTAMP_NTZ
) EXTERNAL_VOLUME = 'TRANSCOM_TSPANNICEBERG_EXTVOL'
CATALOG = 'SNOWFLAKE'
BASE_LOCATION = 'noaaweather/';