diff --git a/demos/nifi-kafka-druid-earthquake-data/IngestEarthquakesToKafka.json b/demos/nifi-kafka-druid-earthquake-data/IngestEarthquakesToKafka.json deleted file mode 100644 index 1e30d720..00000000 --- a/demos/nifi-kafka-druid-earthquake-data/IngestEarthquakesToKafka.json +++ /dev/null @@ -1 +0,0 @@ -{ "flowContents": { "identifier": "a1618dd5-e11a-3ff4-9a6c-62607956b4f4", "instanceIdentifier": "a52fc536-0195-1000-ffff-ffffb77a46e7", "name": "IngestEarthquakesToKafka_raw", "comments": "", "position": { "x": -408.0, "y": -176.0 }, "processGroups": [], "remoteProcessGroups": [], "processors": [ { "identifier": "6f74b992-9f25-3fa8-ada3-eebe335159b3", "instanceIdentifier": "a5301fb3-0195-1000-0000-0000734e0ae7", "name": "SplitRecord", "comments": "", "position": { "x": 472.0, "y": 376.0 }, "type": "org.apache.nifi.processors.standard.SplitRecord", "bundle": { "group": "org.apache.nifi", "artifact": "nifi-standard-nar", "version": "2.6.0" }, "properties": { "Record Writer": "5e81a80d-728d-3c87-800e-aa957618da55", "Record Reader": "d082021f-5f72-322a-8e1d-c5990d04b2b1", "Records Per Split": "2000" }, "propertyDescriptors": { "Record Writer": { "name": "Record Writer", "displayName": "Record Writer", "identifiesControllerService": true, "sensitive": false, "dynamic": false }, "Record Reader": { "name": "Record Reader", "displayName": "Record Reader", "identifiesControllerService": true, "sensitive": false, "dynamic": false }, "Records Per Split": { "name": "Records Per Split", "displayName": "Records Per Split", "identifiesControllerService": false, "sensitive": false, "dynamic": false } }, "style": {}, "schedulingPeriod": "0 sec", "schedulingStrategy": "TIMER_DRIVEN", "executionNode": "ALL", "penaltyDuration": "30 sec", "yieldDuration": "1 sec", "bulletinLevel": "WARN", "runDurationMillis": 0, "concurrentlySchedulableTaskCount": 1, "autoTerminatedRelationships": [ "original", "failure" ], "scheduledState": "ENABLED", "retryCount": 10, "retriedRelationships": [], "backoffMechanism": "PENALIZE_FLOWFILE", "maxBackoffPeriod": "10 mins", "componentType": "PROCESSOR", "groupIdentifier": "a1618dd5-e11a-3ff4-9a6c-62607956b4f4" }, { "identifier": "e29d8c86-db41-3f3e-9ba7-bffeaefcade6", "instanceIdentifier": "d86d6b2e-ed8f-3d61-9200-5c02c0918eec", "name": "InvokeHTTP", "comments": "", "position": { "x": 473.00000076287904, "y": 142.99998844850802 }, "type": "org.apache.nifi.processors.standard.InvokeHTTP", "bundle": { "group": "org.apache.nifi", "artifact": "nifi-standard-nar", "version": "2.6.0" }, "properties": { "Request Content-Encoding": "DISABLED", "proxy-configuration-service": null, "Request Multipart Form-Data Filename Enabled": "true", "Request Chunked Transfer-Encoding Enabled": "false", "Response Header Request Attributes Prefix": null, "HTTP/2 Disabled": "False", "Connection Timeout": "5 secs", "Response Cookie Strategy": "DISABLED", "Socket Read Timeout": "15 secs", "Socket Idle Connections": "5", "Request Body Enabled": "true", "HTTP URL": "https://repo.stackable.tech/repository/misc/earthquake-data/earthquakes_1950_to_2022.csv", "Request OAuth2 Access Token Provider": null, "Socket Idle Timeout": "5 mins", "Response Redirects Enabled": "True", "Socket Write Timeout": "15 secs", "Request Header Attributes Pattern": null, "Response FlowFile Naming Strategy": "RANDOM", "Response Cache Enabled": "true", "Request Date Header Enabled": "True", "Request Failure Penalization Enabled": "false", "Response Body Attribute Size": "256", "SSL Context Service": null, "Response Generation Required": "false", "Request User-Agent": null, "Response Header Request Attributes Enabled": "false", "HTTP Method": "GET", "Request Username": null, "Request Content-Type": "${mime.type}", "Response Body Attribute Name": null, "Request Digest Authentication Enabled": "false", "Request Multipart Form-Data Name": null, "Response Cache Size": "10MB", "Response Body Ignored": "false" }, "propertyDescriptors": { "Request Content-Encoding": { "name": "Request Content-Encoding", "displayName": "Request Content-Encoding", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "proxy-configuration-service": { "name": "proxy-configuration-service", "displayName": "Proxy Configuration Service", "identifiesControllerService": true, "sensitive": false, "dynamic": false }, "Request Multipart Form-Data Filename Enabled": { "name": "Request Multipart Form-Data Filename Enabled", "displayName": "Request Multipart Form-Data Filename Enabled", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Request Chunked Transfer-Encoding Enabled": { "name": "Request Chunked Transfer-Encoding Enabled", "displayName": "Request Chunked Transfer-Encoding Enabled", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Response Header Request Attributes Prefix": { "name": "Response Header Request Attributes Prefix", "displayName": "Response Header Request Attributes Prefix", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "HTTP/2 Disabled": { "name": "HTTP/2 Disabled", "displayName": "HTTP/2 Disabled", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Connection Timeout": { "name": "Connection Timeout", "displayName": "Connection Timeout", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Response Cookie Strategy": { "name": "Response Cookie Strategy", "displayName": "Response Cookie Strategy", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Request Password": { "name": "Request Password", "displayName": "Request Password", "identifiesControllerService": false, "sensitive": true, "dynamic": false }, "Socket Read Timeout": { "name": "Socket Read Timeout", "displayName": "Socket Read Timeout", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Socket Idle Connections": { "name": "Socket Idle Connections", "displayName": "Socket Idle Connections", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Request Body Enabled": { "name": "Request Body Enabled", "displayName": "Request Body Enabled", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "HTTP URL": { "name": "HTTP URL", "displayName": "HTTP URL", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Request OAuth2 Access Token Provider": { "name": "Request OAuth2 Access Token Provider", "displayName": "Request OAuth2 Access Token Provider", "identifiesControllerService": true, "sensitive": false, "dynamic": false }, "Socket Idle Timeout": { "name": "Socket Idle Timeout", "displayName": "Socket Idle Timeout", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Response Redirects Enabled": { "name": "Response Redirects Enabled", "displayName": "Response Redirects Enabled", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Socket Write Timeout": { "name": "Socket Write Timeout", "displayName": "Socket Write Timeout", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Request Header Attributes Pattern": { "name": "Request Header Attributes Pattern", "displayName": "Request Header Attributes Pattern", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Response FlowFile Naming Strategy": { "name": "Response FlowFile Naming Strategy", "displayName": "Response FlowFile Naming Strategy", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Response Cache Enabled": { "name": "Response Cache Enabled", "displayName": "Response Cache Enabled", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Request Date Header Enabled": { "name": "Request Date Header Enabled", "displayName": "Request Date Header Enabled", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Request Failure Penalization Enabled": { "name": "Request Failure Penalization Enabled", "displayName": "Request Failure Penalization Enabled", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Response Body Attribute Size": { "name": "Response Body Attribute Size", "displayName": "Response Body Attribute Size", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "SSL Context Service": { "name": "SSL Context Service", "displayName": "SSL Context Service", "identifiesControllerService": true, "sensitive": false, "dynamic": false }, "Response Generation Required": { "name": "Response Generation Required", "displayName": "Response Generation Required", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Request User-Agent": { "name": "Request User-Agent", "displayName": "Request User-Agent", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Response Header Request Attributes Enabled": { "name": "Response Header Request Attributes Enabled", "displayName": "Response Header Request Attributes Enabled", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "HTTP Method": { "name": "HTTP Method", "displayName": "HTTP Method", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Request Username": { "name": "Request Username", "displayName": "Request Username", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Request Content-Type": { "name": "Request Content-Type", "displayName": "Request Content-Type", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Response Body Attribute Name": { "name": "Response Body Attribute Name", "displayName": "Response Body Attribute Name", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Request Digest Authentication Enabled": { "name": "Request Digest Authentication Enabled", "displayName": "Request Digest Authentication Enabled", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Request Multipart Form-Data Name": { "name": "Request Multipart Form-Data Name", "displayName": "Request Multipart Form-Data Name", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Response Cache Size": { "name": "Response Cache Size", "displayName": "Response Cache Size", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Response Body Ignored": { "name": "Response Body Ignored", "displayName": "Response Body Ignored", "identifiesControllerService": false, "sensitive": false, "dynamic": false } }, "style": {}, "schedulingPeriod": "365000 days", "schedulingStrategy": "TIMER_DRIVEN", "executionNode": "PRIMARY", "penaltyDuration": "30 sec", "yieldDuration": "1 sec", "bulletinLevel": "WARN", "runDurationMillis": 0, "concurrentlySchedulableTaskCount": 1, "autoTerminatedRelationships": [ "No Retry", "Retry", "Original", "Failure" ], "scheduledState": "ENABLED", "retryCount": 10, "retriedRelationships": [], "backoffMechanism": "PENALIZE_FLOWFILE", "maxBackoffPeriod": "10 mins", "componentType": "PROCESSOR", "groupIdentifier": "a1618dd5-e11a-3ff4-9a6c-62607956b4f4" }, { "identifier": "cd097bd8-d55f-3685-b63f-56d46d14c251", "instanceIdentifier": "2ab64ebf-3492-348b-8b9a-7a97e920220f", "name": "PublishKafka", "comments": "", "position": { "x": 472.0, "y": 608.0 }, "type": "org.apache.nifi.kafka.processors.PublishKafka", "bundle": { "group": "org.apache.nifi", "artifact": "nifi-kafka-nar", "version": "2.6.0" }, "properties": { "compression.type": "none", "acks": "all", "Message Demarcator": null, "Kafka Key Attribute Encoding": "utf-8", "Record Reader": "d082021f-5f72-322a-8e1d-c5990d04b2b1", "Record Metadata Strategy": "FROM_PROPERTIES", "Transactional ID Prefix": null, "Header Encoding": "UTF-8", "max.request.size": "1 MB", "Topic Name": "earthquakes", "Kafka Key": null, "partition": null, "Kafka Connection Service": "db258647-11aa-30c5-a8c6-a49a4296cf19", "Publish Strategy": "USE_VALUE", "Record Key Writer": null, "Failure Strategy": "Route to Failure", "partitioner.class": "org.apache.kafka.clients.producer.internals.DefaultPartitioner", "Record Writer": "f3254a5e-0024-31c3-b59a-9edff2a099c9", "Transactions Enabled": "false", "Message Key Field": null, "FlowFile Attribute Header Pattern": null }, "propertyDescriptors": { "compression.type": { "name": "compression.type", "displayName": "Compression Type", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "acks": { "name": "acks", "displayName": "Delivery Guarantee", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Message Demarcator": { "name": "Message Demarcator", "displayName": "Message Demarcator", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Kafka Key Attribute Encoding": { "name": "Kafka Key Attribute Encoding", "displayName": "Kafka Key Attribute Encoding", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Record Reader": { "name": "Record Reader", "displayName": "Record Reader", "identifiesControllerService": true, "sensitive": false, "dynamic": false }, "Record Metadata Strategy": { "name": "Record Metadata Strategy", "displayName": "Record Metadata Strategy", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Transactional ID Prefix": { "name": "Transactional ID Prefix", "displayName": "Transactional ID Prefix", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Header Encoding": { "name": "Header Encoding", "displayName": "Header Encoding", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "max.request.size": { "name": "max.request.size", "displayName": "Max Request Size", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Topic Name": { "name": "Topic Name", "displayName": "Topic Name", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Kafka Key": { "name": "Kafka Key", "displayName": "Kafka Key", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "partition": { "name": "partition", "displayName": "Partition", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Kafka Connection Service": { "name": "Kafka Connection Service", "displayName": "Kafka Connection Service", "identifiesControllerService": true, "sensitive": false, "dynamic": false }, "Publish Strategy": { "name": "Publish Strategy", "displayName": "Publish Strategy", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Record Key Writer": { "name": "Record Key Writer", "displayName": "Record Key Writer", "identifiesControllerService": true, "sensitive": false, "dynamic": false }, "Failure Strategy": { "name": "Failure Strategy", "displayName": "Failure Strategy", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "partitioner.class": { "name": "partitioner.class", "displayName": "Partitioner Class", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Record Writer": { "name": "Record Writer", "displayName": "Record Writer", "identifiesControllerService": true, "sensitive": false, "dynamic": false }, "Transactions Enabled": { "name": "Transactions Enabled", "displayName": "Transactions Enabled", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Message Key Field": { "name": "Message Key Field", "displayName": "Message Key Field", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "FlowFile Attribute Header Pattern": { "name": "FlowFile Attribute Header Pattern", "displayName": "FlowFile Attribute Header Pattern", "identifiesControllerService": false, "sensitive": false, "dynamic": false } }, "style": {}, "schedulingPeriod": "0 sec", "schedulingStrategy": "TIMER_DRIVEN", "executionNode": "ALL", "penaltyDuration": "30 sec", "yieldDuration": "1 sec", "bulletinLevel": "WARN", "runDurationMillis": 0, "concurrentlySchedulableTaskCount": 1, "autoTerminatedRelationships": [ "success", "failure" ], "scheduledState": "ENABLED", "retryCount": 10, "retriedRelationships": [], "backoffMechanism": "PENALIZE_FLOWFILE", "maxBackoffPeriod": "10 mins", "componentType": "PROCESSOR", "groupIdentifier": "a1618dd5-e11a-3ff4-9a6c-62607956b4f4" } ], "inputPorts": [], "outputPorts": [], "connections": [ { "identifier": "4a16d5e1-b4fa-3018-ba09-504a16238421", "instanceIdentifier": "64b996e2-f357-34fe-aee6-e24d6fd201e4", "name": "", "source": { "id": "e29d8c86-db41-3f3e-9ba7-bffeaefcade6", "type": "PROCESSOR", "groupId": "a1618dd5-e11a-3ff4-9a6c-62607956b4f4", "name": "InvokeHTTP", "comments": "", "instanceIdentifier": "d86d6b2e-ed8f-3d61-9200-5c02c0918eec" }, "destination": { "id": "6f74b992-9f25-3fa8-ada3-eebe335159b3", "type": "PROCESSOR", "groupId": "a1618dd5-e11a-3ff4-9a6c-62607956b4f4", "name": "SplitRecord", "comments": "", "instanceIdentifier": "a5301fb3-0195-1000-0000-0000734e0ae7" }, "labelIndex": 1, "zIndex": 0, "selectedRelationships": [ "Response" ], "backPressureObjectThreshold": 10000, "backPressureDataSizeThreshold": "1 GB", "flowFileExpiration": "0 sec", "prioritizers": [], "bends": [], "loadBalanceStrategy": "DO_NOT_LOAD_BALANCE", "partitioningAttribute": "", "loadBalanceCompression": "DO_NOT_COMPRESS", "componentType": "CONNECTION", "groupIdentifier": "a1618dd5-e11a-3ff4-9a6c-62607956b4f4" }, { "identifier": "1025098e-3c22-37c0-9560-203bc51ef427", "instanceIdentifier": "a530ccd7-0195-1000-0000-00004ed00e40", "name": "", "source": { "id": "6f74b992-9f25-3fa8-ada3-eebe335159b3", "type": "PROCESSOR", "groupId": "a1618dd5-e11a-3ff4-9a6c-62607956b4f4", "name": "SplitRecord", "comments": "", "instanceIdentifier": "a5301fb3-0195-1000-0000-0000734e0ae7" }, "destination": { "id": "cd097bd8-d55f-3685-b63f-56d46d14c251", "type": "PROCESSOR", "groupId": "a1618dd5-e11a-3ff4-9a6c-62607956b4f4", "name": "PublishKafka", "comments": "", "instanceIdentifier": "2ab64ebf-3492-348b-8b9a-7a97e920220f" }, "labelIndex": 0, "zIndex": 0, "selectedRelationships": [ "splits" ], "backPressureObjectThreshold": 10000, "backPressureDataSizeThreshold": "1 GB", "flowFileExpiration": "0 sec", "prioritizers": [], "bends": [], "loadBalanceStrategy": "DO_NOT_LOAD_BALANCE", "partitioningAttribute": "", "loadBalanceCompression": "DO_NOT_COMPRESS", "componentType": "CONNECTION", "groupIdentifier": "a1618dd5-e11a-3ff4-9a6c-62607956b4f4" } ], "labels": [], "funnels": [], "controllerServices": [ { "identifier": "db258647-11aa-30c5-a8c6-a49a4296cf19", "instanceIdentifier": "d22c6c6b-4cd7-354f-b5d4-ba93b975ec64", "name": "Kafka3ConnectionService", "comments": "", "type": "org.apache.nifi.kafka.service.Kafka3ConnectionService", "bundle": { "group": "org.apache.nifi", "artifact": "nifi-kafka-3-service-nar", "version": "2.6.0" }, "properties": { "security.protocol": "SSL", "SSL Context Service": "7fce889c-fc16-3f69-b7b4-44a90a81ae38", "ack.wait.time": "5 sec", "sasl.username": null, "max.poll.records": "10000", "sasl.mechanism": "GSSAPI", "isolation.level": "read_committed", "max.block.ms": "5 sec", "bootstrap.servers": "kafka-broker-default.default.svc.cluster.local:9093", "kerberos-user-service": null, "sasl.kerberos.service.name": null, "default.api.timeout.ms": "60 sec" }, "propertyDescriptors": { "ack.wait.time": { "name": "ack.wait.time", "displayName": "Acknowledgment Wait Time", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "sasl.username": { "name": "sasl.username", "displayName": "SASL Username", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "max.poll.records": { "name": "max.poll.records", "displayName": "Max Poll Records", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "bootstrap.servers": { "name": "bootstrap.servers", "displayName": "Bootstrap Servers", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "sasl.kerberos.service.name": { "name": "sasl.kerberos.service.name", "displayName": "Kerberos Service Name", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "security.protocol": { "name": "security.protocol", "displayName": "Security Protocol", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "SSL Context Service": { "name": "SSL Context Service", "displayName": "SSL Context Service", "identifiesControllerService": true, "sensitive": false, "dynamic": false }, "sasl.mechanism": { "name": "sasl.mechanism", "displayName": "SASL Mechanism", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "isolation.level": { "name": "isolation.level", "displayName": "Transaction Isolation Level", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "max.block.ms": { "name": "max.block.ms", "displayName": "Max Metadata Wait Time", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "kerberos-user-service": { "name": "kerberos-user-service", "displayName": "Kerberos User Service", "identifiesControllerService": true, "sensitive": false, "dynamic": false }, "sasl.password": { "name": "sasl.password", "displayName": "SASL Password", "identifiesControllerService": false, "sensitive": true, "dynamic": false }, "default.api.timeout.ms": { "name": "default.api.timeout.ms", "displayName": "Client Timeout", "identifiesControllerService": false, "sensitive": false, "dynamic": false } }, "controllerServiceApis": [ { "type": "org.apache.nifi.kafka.service.api.KafkaConnectionService", "bundle": { "group": "org.apache.nifi", "artifact": "nifi-kafka-service-api-nar", "version": "2.6.0" } } ], "scheduledState": "DISABLED", "bulletinLevel": "WARN", "componentType": "CONTROLLER_SERVICE", "groupIdentifier": "a1618dd5-e11a-3ff4-9a6c-62607956b4f4" }, { "identifier": "d082021f-5f72-322a-8e1d-c5990d04b2b1", "instanceIdentifier": "ab119a98-0d96-3556-b54f-411fd9cdd462", "name": "CSVReader", "type": "org.apache.nifi.csv.CSVReader", "bundle": { "group": "org.apache.nifi", "artifact": "nifi-record-serialization-services-nar", "version": "2.6.0" }, "properties": { "schema-reference-reader": null, "ignore-csv-header": "false", "schema-branch": null, "Trim double quote": "true", "CSV Format": "custom", "Quote Character": "\"", "csvutils-allow-duplicate-header-names": "true", "Value Separator": ",", "Record Separator": "\\n", "Timestamp Format": null, "Escape Character": "\\", "Date Format": null, "Null String": null, "Skip Header Line": "false", "Trim Fields": "true", "schema-name": "${schema.name}", "schema-registry": null, "csv-reader-csv-parser": "commons-csv", "Time Format": null, "Comment Marker": null, "schema-access-strategy": "infer-schema", "schema-version": null, "csvutils-character-set": "UTF-8", "schema-text": "${avro.schema}" }, "propertyDescriptors": { "schema-reference-reader": { "name": "schema-reference-reader", "displayName": "Schema Reference Reader", "identifiesControllerService": true, "sensitive": false, "dynamic": false }, "ignore-csv-header": { "name": "ignore-csv-header", "displayName": "Ignore CSV Header Column Names", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "schema-branch": { "name": "schema-branch", "displayName": "Schema Branch", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Trim double quote": { "name": "Trim double quote", "displayName": "Trim double quote", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "CSV Format": { "name": "CSV Format", "displayName": "CSV Format", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Quote Character": { "name": "Quote Character", "displayName": "Quote Character", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "csvutils-allow-duplicate-header-names": { "name": "csvutils-allow-duplicate-header-names", "displayName": "Allow Duplicate Header Names", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Value Separator": { "name": "Value Separator", "displayName": "Value Separator", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Record Separator": { "name": "Record Separator", "displayName": "Record Separator", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Timestamp Format": { "name": "Timestamp Format", "displayName": "Timestamp Format", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Escape Character": { "name": "Escape Character", "displayName": "Escape Character", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Date Format": { "name": "Date Format", "displayName": "Date Format", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Null String": { "name": "Null String", "displayName": "Null String", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Skip Header Line": { "name": "Skip Header Line", "displayName": "Treat First Line as Header", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Trim Fields": { "name": "Trim Fields", "displayName": "Trim Fields", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "schema-name": { "name": "schema-name", "displayName": "Schema Name", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "schema-registry": { "name": "schema-registry", "displayName": "Schema Registry", "identifiesControllerService": true, "sensitive": false, "dynamic": false }, "csv-reader-csv-parser": { "name": "csv-reader-csv-parser", "displayName": "CSV Parser", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Time Format": { "name": "Time Format", "displayName": "Time Format", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Comment Marker": { "name": "Comment Marker", "displayName": "Comment Marker", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "schema-access-strategy": { "name": "schema-access-strategy", "displayName": "Schema Access Strategy", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "schema-version": { "name": "schema-version", "displayName": "Schema Version", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "csvutils-character-set": { "name": "csvutils-character-set", "displayName": "Character Set", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "schema-text": { "name": "schema-text", "displayName": "Schema Text", "identifiesControllerService": false, "sensitive": false, "dynamic": false } }, "controllerServiceApis": [ { "type": "org.apache.nifi.serialization.RecordReaderFactory", "bundle": { "group": "org.apache.nifi", "artifact": "nifi-standard-services-api-nar", "version": "2.6.0" } } ], "scheduledState": "DISABLED", "bulletinLevel": "WARN", "componentType": "CONTROLLER_SERVICE", "groupIdentifier": "a1618dd5-e11a-3ff4-9a6c-62607956b4f4" }, { "identifier": "7fce889c-fc16-3f69-b7b4-44a90a81ae38", "instanceIdentifier": "3238df43-ee04-3bec-aa9b-13efe7c0f72a", "name": "StandardRestrictedSSLContextService", "comments": "", "type": "org.apache.nifi.ssl.StandardRestrictedSSLContextService", "bundle": { "group": "org.apache.nifi", "artifact": "nifi-ssl-context-service-nar", "version": "2.6.0" }, "properties": { "Truststore Type": "PKCS12", "Truststore Password": "secret", "SSL Protocol": "TLSv1.3", "Keystore Type": "PKCS12", "Keystore Password": "secret", "Truststore Filename": "/stackable/keystore/truststore.p12", "Keystore Filename": "/stackable/keystore/keystore.p12" }, "propertyDescriptors": { "Truststore Type": { "name": "Truststore Type", "displayName": "Truststore Type", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "SSL Protocol": { "name": "SSL Protocol", "displayName": "TLS Protocol", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Keystore Type": { "name": "Keystore Type", "displayName": "Keystore Type", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Truststore Filename": { "name": "Truststore Filename", "displayName": "Truststore Filename", "identifiesControllerService": false, "sensitive": false, "dynamic": false, "resourceDefinition": { "cardinality": "SINGLE", "resourceTypes": [ "FILE" ] } }, "Keystore Password": { "name": "Keystore Password", "displayName": "Keystore Password", "identifiesControllerService": false, "sensitive": true, "dynamic": false }, "key-password": { "name": "key-password", "displayName": "Key Password", "identifiesControllerService": false, "sensitive": true, "dynamic": false }, "Truststore Password": { "name": "Truststore Password", "displayName": "Truststore Password", "identifiesControllerService": false, "sensitive": true, "dynamic": false }, "Keystore Filename": { "name": "Keystore Filename", "displayName": "Keystore Filename", "identifiesControllerService": false, "sensitive": false, "dynamic": false, "resourceDefinition": { "cardinality": "SINGLE", "resourceTypes": [ "FILE" ] } } }, "controllerServiceApis": [ { "type": "org.apache.nifi.ssl.SSLContextService", "bundle": { "group": "org.apache.nifi", "artifact": "nifi-standard-services-api-nar", "version": "2.6.0" } }, { "type": "org.apache.nifi.ssl.RestrictedSSLContextService", "bundle": { "group": "org.apache.nifi", "artifact": "nifi-standard-services-api-nar", "version": "2.6.0" } }, { "type": "org.apache.nifi.ssl.SSLContextProvider", "bundle": { "group": "org.apache.nifi", "artifact": "nifi-standard-services-api-nar", "version": "2.6.0" } } ], "scheduledState": "DISABLED", "bulletinLevel": "WARN", "componentType": "CONTROLLER_SERVICE", "groupIdentifier": "a1618dd5-e11a-3ff4-9a6c-62607956b4f4" }, { "identifier": "f3254a5e-0024-31c3-b59a-9edff2a099c9", "instanceIdentifier": "8dd599ec-3c90-3a34-8213-185a6d1bfb92", "name": "JsonRecordSetWriter", "type": "org.apache.nifi.json.JsonRecordSetWriter", "bundle": { "group": "org.apache.nifi", "artifact": "nifi-record-serialization-services-nar", "version": "2.6.0" }, "properties": { "schema-reference-reader": null, "schema-branch": null, "Allow Scientific Notation": "false", "compression-level": "1", "schema-cache": null, "Timestamp Format": null, "Date Format": null, "Pretty Print JSON": "false", "compression-format": "none", "Schema Write Strategy": "no-schema", "suppress-nulls": "never-suppress", "output-grouping": "output-array", "schema-name": "${schema.name}", "schema-registry": null, "Time Format": null, "schema-access-strategy": "inherit-record-schema", "schema-version": null, "schema-text": "${avro.schema}", "Schema Reference Writer": null }, "propertyDescriptors": { "schema-reference-reader": { "name": "schema-reference-reader", "displayName": "Schema Reference Reader", "identifiesControllerService": true, "sensitive": false, "dynamic": false }, "schema-branch": { "name": "schema-branch", "displayName": "Schema Branch", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Allow Scientific Notation": { "name": "Allow Scientific Notation", "displayName": "Allow Scientific Notation", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "compression-level": { "name": "compression-level", "displayName": "Compression Level", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "schema-cache": { "name": "schema-cache", "displayName": "Schema Cache", "identifiesControllerService": true, "sensitive": false, "dynamic": false }, "Timestamp Format": { "name": "Timestamp Format", "displayName": "Timestamp Format", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Date Format": { "name": "Date Format", "displayName": "Date Format", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Pretty Print JSON": { "name": "Pretty Print JSON", "displayName": "Pretty Print JSON", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "compression-format": { "name": "compression-format", "displayName": "Compression Format", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Schema Write Strategy": { "name": "Schema Write Strategy", "displayName": "Schema Write Strategy", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "suppress-nulls": { "name": "suppress-nulls", "displayName": "Suppress Null Values", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "output-grouping": { "name": "output-grouping", "displayName": "Output Grouping", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "schema-name": { "name": "schema-name", "displayName": "Schema Name", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "schema-registry": { "name": "schema-registry", "displayName": "Schema Registry", "identifiesControllerService": true, "sensitive": false, "dynamic": false }, "Time Format": { "name": "Time Format", "displayName": "Time Format", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "schema-access-strategy": { "name": "schema-access-strategy", "displayName": "Schema Access Strategy", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "schema-version": { "name": "schema-version", "displayName": "Schema Version", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "schema-text": { "name": "schema-text", "displayName": "Schema Text", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Schema Reference Writer": { "name": "Schema Reference Writer", "displayName": "Schema Reference Writer", "identifiesControllerService": true, "sensitive": false, "dynamic": false } }, "controllerServiceApis": [ { "type": "org.apache.nifi.serialization.RecordSetWriterFactory", "bundle": { "group": "org.apache.nifi", "artifact": "nifi-standard-services-api-nar", "version": "2.6.0" } } ], "scheduledState": "DISABLED", "bulletinLevel": "WARN", "componentType": "CONTROLLER_SERVICE", "groupIdentifier": "a1618dd5-e11a-3ff4-9a6c-62607956b4f4" }, { "identifier": "5e81a80d-728d-3c87-800e-aa957618da55", "instanceIdentifier": "a53045c6-0195-1000-ffff-ffffef6c13db", "name": "CSVRecordSetWriter", "comments": "", "type": "org.apache.nifi.csv.CSVRecordSetWriter", "bundle": { "group": "org.apache.nifi", "artifact": "nifi-record-serialization-services-nar", "version": "2.6.0" }, "properties": { "schema-reference-reader": null, "schema-branch": null, "CSV Format": "custom", "Include Header Line": "true", "schema-cache": null, "Escape Character": "\\", "Date Format": null, "Null String": null, "Trim Fields": "true", "schema-registry": null, "Time Format": null, "Comment Marker": null, "schema-access-strategy": "inherit-record-schema", "schema-version": null, "csvutils-character-set": "UTF-8", "schema-text": "${avro.schema}", "csv-writer": "commons-csv", "Quote Character": "\"", "Value Separator": ",", "Record Separator": "\\n", "Timestamp Format": null, "Schema Write Strategy": "no-schema", "schema-name": "${schema.name}", "Quote Mode": "MINIMAL", "Schema Reference Writer": null, "Include Trailing Delimiter": "false" }, "propertyDescriptors": { "schema-reference-reader": { "name": "schema-reference-reader", "displayName": "Schema Reference Reader", "identifiesControllerService": true, "sensitive": false, "dynamic": false }, "schema-branch": { "name": "schema-branch", "displayName": "Schema Branch", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "CSV Format": { "name": "CSV Format", "displayName": "CSV Format", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Include Header Line": { "name": "Include Header Line", "displayName": "Include Header Line", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "schema-cache": { "name": "schema-cache", "displayName": "Schema Cache", "identifiesControllerService": true, "sensitive": false, "dynamic": false }, "Escape Character": { "name": "Escape Character", "displayName": "Escape Character", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Date Format": { "name": "Date Format", "displayName": "Date Format", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Null String": { "name": "Null String", "displayName": "Null String", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Trim Fields": { "name": "Trim Fields", "displayName": "Trim Fields", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "schema-registry": { "name": "schema-registry", "displayName": "Schema Registry", "identifiesControllerService": true, "sensitive": false, "dynamic": false }, "Time Format": { "name": "Time Format", "displayName": "Time Format", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Comment Marker": { "name": "Comment Marker", "displayName": "Comment Marker", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "schema-access-strategy": { "name": "schema-access-strategy", "displayName": "Schema Access Strategy", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "schema-version": { "name": "schema-version", "displayName": "Schema Version", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "csvutils-character-set": { "name": "csvutils-character-set", "displayName": "Character Set", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "schema-text": { "name": "schema-text", "displayName": "Schema Text", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "csv-writer": { "name": "csv-writer", "displayName": "CSV Writer", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Quote Character": { "name": "Quote Character", "displayName": "Quote Character", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Value Separator": { "name": "Value Separator", "displayName": "Value Separator", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Record Separator": { "name": "Record Separator", "displayName": "Record Separator", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Timestamp Format": { "name": "Timestamp Format", "displayName": "Timestamp Format", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Schema Write Strategy": { "name": "Schema Write Strategy", "displayName": "Schema Write Strategy", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "schema-name": { "name": "schema-name", "displayName": "Schema Name", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Quote Mode": { "name": "Quote Mode", "displayName": "Quote Mode", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Schema Reference Writer": { "name": "Schema Reference Writer", "displayName": "Schema Reference Writer", "identifiesControllerService": true, "sensitive": false, "dynamic": false }, "Include Trailing Delimiter": { "name": "Include Trailing Delimiter", "displayName": "Include Trailing Delimiter", "identifiesControllerService": false, "sensitive": false, "dynamic": false } }, "controllerServiceApis": [ { "type": "org.apache.nifi.serialization.RecordSetWriterFactory", "bundle": { "group": "org.apache.nifi", "artifact": "nifi-standard-services-api-nar", "version": "2.6.0" } } ], "scheduledState": "DISABLED", "bulletinLevel": "WARN", "componentType": "CONTROLLER_SERVICE", "groupIdentifier": "a1618dd5-e11a-3ff4-9a6c-62607956b4f4" } ], "defaultFlowFileExpiration": "0 sec", "defaultBackPressureObjectThreshold": 10000, "defaultBackPressureDataSizeThreshold": "1 GB", "scheduledState": "ENABLED", "executionEngine": "INHERITED", "maxConcurrentTasks": 1, "statelessFlowTimeout": "1 min", "flowFileConcurrency": "UNBOUNDED", "flowFileOutboundPolicy": "STREAM_WHEN_AVAILABLE", "componentType": "PROCESS_GROUP" }, "externalControllerServices": {}, "parameterContexts": {}, "flowEncodingVersion": "1.0", "parameterProviders": {}, "latest": false } diff --git a/demos/nifi-kafka-druid-earthquake-data/create-nifi-ingestion-job.yaml b/demos/nifi-kafka-druid-earthquake-data/create-nifi-ingestion-job.yaml index 306e42cf..c59b2698 100644 --- a/demos/nifi-kafka-druid-earthquake-data/create-nifi-ingestion-job.yaml +++ b/demos/nifi-kafka-druid-earthquake-data/create-nifi-ingestion-job.yaml @@ -32,6 +32,8 @@ spec: volumeMounts: - name: script mountPath: /tmp/script + - name: flow + mountPath: /tmp/flow - name: nifi-admin-credentials-secret mountPath: /nifi-admin-credentials-secret env: @@ -43,6 +45,9 @@ spec: - name: script configMap: name: create-nifi-ingestion-job-script + - name: flow + configMap: + name: ingest-earthquake-data-flow - name: nifi-admin-credentials-secret secret: secretName: nifi-admin-credentials-secret @@ -76,11 +81,6 @@ data: service_login(username=USERNAME, password=PASSWORD) print("Logged in") - response = requests.get("https://raw.githubusercontent.com/stackabletech/demos/refs/heads/main/demos/nifi-kafka-druid-earthquake-data/IngestEarthquakesToKafka.json") - filename = "/tmp/IngestEarthquakesToKafka.json" - with open(filename, "wb") as f: - f.write(response.content) - pg_id = get_root_pg_id() if not nipyapi.config.nifi_config.api_client: @@ -102,7 +102,7 @@ data: ('clientId', nipyapi.nifi.FlowApi().generate_client_id()), ], files={ - 'file': filename + 'file': "/tmp/flow/IngestEarthquakesToKafka.json" }, auth_settings=['tokenAuth']) @@ -122,3 +122,12 @@ data: print(f"Failed to schedule controller {controller.component.name}: {e}") schedule_process_group(pg_id, scheduled=True) +--- +apiVersion: v1 +kind: ConfigMap +metadata: + name: ingest-earthquake-data-flow + namespace: default +data: + IngestEarthquakesToKafka.json: | + { "flowContents": { "identifier": "a1618dd5-e11a-3ff4-9a6c-62607956b4f4", "instanceIdentifier": "a52fc536-0195-1000-ffff-ffffb77a46e7", "name": "IngestEarthquakesToKafka_raw", "comments": "", "position": { "x": -408.0, "y": -176.0 }, "processGroups": [], "remoteProcessGroups": [], "processors": [ { "identifier": "6f74b992-9f25-3fa8-ada3-eebe335159b3", "instanceIdentifier": "a5301fb3-0195-1000-0000-0000734e0ae7", "name": "SplitRecord", "comments": "", "position": { "x": 472.0, "y": 376.0 }, "type": "org.apache.nifi.processors.standard.SplitRecord", "bundle": { "group": "org.apache.nifi", "artifact": "nifi-standard-nar", "version": "2.6.0" }, "properties": { "Record Writer": "5e81a80d-728d-3c87-800e-aa957618da55", "Record Reader": "d082021f-5f72-322a-8e1d-c5990d04b2b1", "Records Per Split": "2000" }, "propertyDescriptors": { "Record Writer": { "name": "Record Writer", "displayName": "Record Writer", "identifiesControllerService": true, "sensitive": false, "dynamic": false }, "Record Reader": { "name": "Record Reader", "displayName": "Record Reader", "identifiesControllerService": true, "sensitive": false, "dynamic": false }, "Records Per Split": { "name": "Records Per Split", "displayName": "Records Per Split", "identifiesControllerService": false, "sensitive": false, "dynamic": false } }, "style": {}, "schedulingPeriod": "0 sec", "schedulingStrategy": "TIMER_DRIVEN", "executionNode": "ALL", "penaltyDuration": "30 sec", "yieldDuration": "1 sec", "bulletinLevel": "WARN", "runDurationMillis": 0, "concurrentlySchedulableTaskCount": 1, "autoTerminatedRelationships": [ "original", "failure" ], "scheduledState": "ENABLED", "retryCount": 10, "retriedRelationships": [], "backoffMechanism": "PENALIZE_FLOWFILE", "maxBackoffPeriod": "10 mins", "componentType": "PROCESSOR", "groupIdentifier": "a1618dd5-e11a-3ff4-9a6c-62607956b4f4" }, { "identifier": "e29d8c86-db41-3f3e-9ba7-bffeaefcade6", "instanceIdentifier": "d86d6b2e-ed8f-3d61-9200-5c02c0918eec", "name": "InvokeHTTP", "comments": "", "position": { "x": 473.00000076287904, "y": 142.99998844850802 }, "type": "org.apache.nifi.processors.standard.InvokeHTTP", "bundle": { "group": "org.apache.nifi", "artifact": "nifi-standard-nar", "version": "2.6.0" }, "properties": { "Request Content-Encoding": "DISABLED", "proxy-configuration-service": null, "Request Multipart Form-Data Filename Enabled": "true", "Request Chunked Transfer-Encoding Enabled": "false", "Response Header Request Attributes Prefix": null, "HTTP/2 Disabled": "False", "Connection Timeout": "5 secs", "Response Cookie Strategy": "DISABLED", "Socket Read Timeout": "15 secs", "Socket Idle Connections": "5", "Request Body Enabled": "true", "HTTP URL": "https://repo.stackable.tech/repository/misc/earthquake-data/earthquakes_1950_to_2022.csv", "Request OAuth2 Access Token Provider": null, "Socket Idle Timeout": "5 mins", "Response Redirects Enabled": "True", "Socket Write Timeout": "15 secs", "Request Header Attributes Pattern": null, "Response FlowFile Naming Strategy": "RANDOM", "Response Cache Enabled": "true", "Request Date Header Enabled": "True", "Request Failure Penalization Enabled": "false", "Response Body Attribute Size": "256", "SSL Context Service": null, "Response Generation Required": "false", "Request User-Agent": null, "Response Header Request Attributes Enabled": "false", "HTTP Method": "GET", "Request Username": null, "Request Content-Type": "${mime.type}", "Response Body Attribute Name": null, "Request Digest Authentication Enabled": "false", "Request Multipart Form-Data Name": null, "Response Cache Size": "10MB", "Response Body Ignored": "false" }, "propertyDescriptors": { "Request Content-Encoding": { "name": "Request Content-Encoding", "displayName": "Request Content-Encoding", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "proxy-configuration-service": { "name": "proxy-configuration-service", "displayName": "Proxy Configuration Service", "identifiesControllerService": true, "sensitive": false, "dynamic": false }, "Request Multipart Form-Data Filename Enabled": { "name": "Request Multipart Form-Data Filename Enabled", "displayName": "Request Multipart Form-Data Filename Enabled", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Request Chunked Transfer-Encoding Enabled": { "name": "Request Chunked Transfer-Encoding Enabled", "displayName": "Request Chunked Transfer-Encoding Enabled", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Response Header Request Attributes Prefix": { "name": "Response Header Request Attributes Prefix", "displayName": "Response Header Request Attributes Prefix", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "HTTP/2 Disabled": { "name": "HTTP/2 Disabled", "displayName": "HTTP/2 Disabled", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Connection Timeout": { "name": "Connection Timeout", "displayName": "Connection Timeout", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Response Cookie Strategy": { "name": "Response Cookie Strategy", "displayName": "Response Cookie Strategy", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Request Password": { "name": "Request Password", "displayName": "Request Password", "identifiesControllerService": false, "sensitive": true, "dynamic": false }, "Socket Read Timeout": { "name": "Socket Read Timeout", "displayName": "Socket Read Timeout", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Socket Idle Connections": { "name": "Socket Idle Connections", "displayName": "Socket Idle Connections", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Request Body Enabled": { "name": "Request Body Enabled", "displayName": "Request Body Enabled", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "HTTP URL": { "name": "HTTP URL", "displayName": "HTTP URL", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Request OAuth2 Access Token Provider": { "name": "Request OAuth2 Access Token Provider", "displayName": "Request OAuth2 Access Token Provider", "identifiesControllerService": true, "sensitive": false, "dynamic": false }, "Socket Idle Timeout": { "name": "Socket Idle Timeout", "displayName": "Socket Idle Timeout", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Response Redirects Enabled": { "name": "Response Redirects Enabled", "displayName": "Response Redirects Enabled", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Socket Write Timeout": { "name": "Socket Write Timeout", "displayName": "Socket Write Timeout", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Request Header Attributes Pattern": { "name": "Request Header Attributes Pattern", "displayName": "Request Header Attributes Pattern", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Response FlowFile Naming Strategy": { "name": "Response FlowFile Naming Strategy", "displayName": "Response FlowFile Naming Strategy", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Response Cache Enabled": { "name": "Response Cache Enabled", "displayName": "Response Cache Enabled", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Request Date Header Enabled": { "name": "Request Date Header Enabled", "displayName": "Request Date Header Enabled", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Request Failure Penalization Enabled": { "name": "Request Failure Penalization Enabled", "displayName": "Request Failure Penalization Enabled", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Response Body Attribute Size": { "name": "Response Body Attribute Size", "displayName": "Response Body Attribute Size", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "SSL Context Service": { "name": "SSL Context Service", "displayName": "SSL Context Service", "identifiesControllerService": true, "sensitive": false, "dynamic": false }, "Response Generation Required": { "name": "Response Generation Required", "displayName": "Response Generation Required", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Request User-Agent": { "name": "Request User-Agent", "displayName": "Request User-Agent", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Response Header Request Attributes Enabled": { "name": "Response Header Request Attributes Enabled", "displayName": "Response Header Request Attributes Enabled", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "HTTP Method": { "name": "HTTP Method", "displayName": "HTTP Method", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Request Username": { "name": "Request Username", "displayName": "Request Username", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Request Content-Type": { "name": "Request Content-Type", "displayName": "Request Content-Type", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Response Body Attribute Name": { "name": "Response Body Attribute Name", "displayName": "Response Body Attribute Name", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Request Digest Authentication Enabled": { "name": "Request Digest Authentication Enabled", "displayName": "Request Digest Authentication Enabled", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Request Multipart Form-Data Name": { "name": "Request Multipart Form-Data Name", "displayName": "Request Multipart Form-Data Name", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Response Cache Size": { "name": "Response Cache Size", "displayName": "Response Cache Size", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Response Body Ignored": { "name": "Response Body Ignored", "displayName": "Response Body Ignored", "identifiesControllerService": false, "sensitive": false, "dynamic": false } }, "style": {}, "schedulingPeriod": "365000 days", "schedulingStrategy": "TIMER_DRIVEN", "executionNode": "PRIMARY", "penaltyDuration": "30 sec", "yieldDuration": "1 sec", "bulletinLevel": "WARN", "runDurationMillis": 0, "concurrentlySchedulableTaskCount": 1, "autoTerminatedRelationships": [ "No Retry", "Retry", "Original", "Failure" ], "scheduledState": "ENABLED", "retryCount": 10, "retriedRelationships": [], "backoffMechanism": "PENALIZE_FLOWFILE", "maxBackoffPeriod": "10 mins", "componentType": "PROCESSOR", "groupIdentifier": "a1618dd5-e11a-3ff4-9a6c-62607956b4f4" }, { "identifier": "cd097bd8-d55f-3685-b63f-56d46d14c251", "instanceIdentifier": "2ab64ebf-3492-348b-8b9a-7a97e920220f", "name": "PublishKafka", "comments": "", "position": { "x": 472.0, "y": 608.0 }, "type": "org.apache.nifi.kafka.processors.PublishKafka", "bundle": { "group": "org.apache.nifi", "artifact": "nifi-kafka-nar", "version": "2.6.0" }, "properties": { "compression.type": "none", "acks": "all", "Message Demarcator": null, "Kafka Key Attribute Encoding": "utf-8", "Record Reader": "d082021f-5f72-322a-8e1d-c5990d04b2b1", "Record Metadata Strategy": "FROM_PROPERTIES", "Transactional ID Prefix": null, "Header Encoding": "UTF-8", "max.request.size": "1 MB", "Topic Name": "earthquakes", "Kafka Key": null, "partition": null, "Kafka Connection Service": "db258647-11aa-30c5-a8c6-a49a4296cf19", "Publish Strategy": "USE_VALUE", "Record Key Writer": null, "Failure Strategy": "Route to Failure", "partitioner.class": "org.apache.kafka.clients.producer.internals.DefaultPartitioner", "Record Writer": "f3254a5e-0024-31c3-b59a-9edff2a099c9", "Transactions Enabled": "false", "Message Key Field": null, "FlowFile Attribute Header Pattern": null }, "propertyDescriptors": { "compression.type": { "name": "compression.type", "displayName": "Compression Type", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "acks": { "name": "acks", "displayName": "Delivery Guarantee", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Message Demarcator": { "name": "Message Demarcator", "displayName": "Message Demarcator", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Kafka Key Attribute Encoding": { "name": "Kafka Key Attribute Encoding", "displayName": "Kafka Key Attribute Encoding", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Record Reader": { "name": "Record Reader", "displayName": "Record Reader", "identifiesControllerService": true, "sensitive": false, "dynamic": false }, "Record Metadata Strategy": { "name": "Record Metadata Strategy", "displayName": "Record Metadata Strategy", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Transactional ID Prefix": { "name": "Transactional ID Prefix", "displayName": "Transactional ID Prefix", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Header Encoding": { "name": "Header Encoding", "displayName": "Header Encoding", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "max.request.size": { "name": "max.request.size", "displayName": "Max Request Size", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Topic Name": { "name": "Topic Name", "displayName": "Topic Name", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Kafka Key": { "name": "Kafka Key", "displayName": "Kafka Key", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "partition": { "name": "partition", "displayName": "Partition", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Kafka Connection Service": { "name": "Kafka Connection Service", "displayName": "Kafka Connection Service", "identifiesControllerService": true, "sensitive": false, "dynamic": false }, "Publish Strategy": { "name": "Publish Strategy", "displayName": "Publish Strategy", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Record Key Writer": { "name": "Record Key Writer", "displayName": "Record Key Writer", "identifiesControllerService": true, "sensitive": false, "dynamic": false }, "Failure Strategy": { "name": "Failure Strategy", "displayName": "Failure Strategy", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "partitioner.class": { "name": "partitioner.class", "displayName": "Partitioner Class", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Record Writer": { "name": "Record Writer", "displayName": "Record Writer", "identifiesControllerService": true, "sensitive": false, "dynamic": false }, "Transactions Enabled": { "name": "Transactions Enabled", "displayName": "Transactions Enabled", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Message Key Field": { "name": "Message Key Field", "displayName": "Message Key Field", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "FlowFile Attribute Header Pattern": { "name": "FlowFile Attribute Header Pattern", "displayName": "FlowFile Attribute Header Pattern", "identifiesControllerService": false, "sensitive": false, "dynamic": false } }, "style": {}, "schedulingPeriod": "0 sec", "schedulingStrategy": "TIMER_DRIVEN", "executionNode": "ALL", "penaltyDuration": "30 sec", "yieldDuration": "1 sec", "bulletinLevel": "WARN", "runDurationMillis": 0, "concurrentlySchedulableTaskCount": 1, "autoTerminatedRelationships": [ "success", "failure" ], "scheduledState": "ENABLED", "retryCount": 10, "retriedRelationships": [], "backoffMechanism": "PENALIZE_FLOWFILE", "maxBackoffPeriod": "10 mins", "componentType": "PROCESSOR", "groupIdentifier": "a1618dd5-e11a-3ff4-9a6c-62607956b4f4" } ], "inputPorts": [], "outputPorts": [], "connections": [ { "identifier": "4a16d5e1-b4fa-3018-ba09-504a16238421", "instanceIdentifier": "64b996e2-f357-34fe-aee6-e24d6fd201e4", "name": "", "source": { "id": "e29d8c86-db41-3f3e-9ba7-bffeaefcade6", "type": "PROCESSOR", "groupId": "a1618dd5-e11a-3ff4-9a6c-62607956b4f4", "name": "InvokeHTTP", "comments": "", "instanceIdentifier": "d86d6b2e-ed8f-3d61-9200-5c02c0918eec" }, "destination": { "id": "6f74b992-9f25-3fa8-ada3-eebe335159b3", "type": "PROCESSOR", "groupId": "a1618dd5-e11a-3ff4-9a6c-62607956b4f4", "name": "SplitRecord", "comments": "", "instanceIdentifier": "a5301fb3-0195-1000-0000-0000734e0ae7" }, "labelIndex": 1, "zIndex": 0, "selectedRelationships": [ "Response" ], "backPressureObjectThreshold": 10000, "backPressureDataSizeThreshold": "1 GB", "flowFileExpiration": "0 sec", "prioritizers": [], "bends": [], "loadBalanceStrategy": "DO_NOT_LOAD_BALANCE", "partitioningAttribute": "", "loadBalanceCompression": "DO_NOT_COMPRESS", "componentType": "CONNECTION", "groupIdentifier": "a1618dd5-e11a-3ff4-9a6c-62607956b4f4" }, { "identifier": "1025098e-3c22-37c0-9560-203bc51ef427", "instanceIdentifier": "a530ccd7-0195-1000-0000-00004ed00e40", "name": "", "source": { "id": "6f74b992-9f25-3fa8-ada3-eebe335159b3", "type": "PROCESSOR", "groupId": "a1618dd5-e11a-3ff4-9a6c-62607956b4f4", "name": "SplitRecord", "comments": "", "instanceIdentifier": "a5301fb3-0195-1000-0000-0000734e0ae7" }, "destination": { "id": "cd097bd8-d55f-3685-b63f-56d46d14c251", "type": "PROCESSOR", "groupId": "a1618dd5-e11a-3ff4-9a6c-62607956b4f4", "name": "PublishKafka", "comments": "", "instanceIdentifier": "2ab64ebf-3492-348b-8b9a-7a97e920220f" }, "labelIndex": 0, "zIndex": 0, "selectedRelationships": [ "splits" ], "backPressureObjectThreshold": 10000, "backPressureDataSizeThreshold": "1 GB", "flowFileExpiration": "0 sec", "prioritizers": [], "bends": [], "loadBalanceStrategy": "DO_NOT_LOAD_BALANCE", "partitioningAttribute": "", "loadBalanceCompression": "DO_NOT_COMPRESS", "componentType": "CONNECTION", "groupIdentifier": "a1618dd5-e11a-3ff4-9a6c-62607956b4f4" } ], "labels": [], "funnels": [], "controllerServices": [ { "identifier": "db258647-11aa-30c5-a8c6-a49a4296cf19", "instanceIdentifier": "d22c6c6b-4cd7-354f-b5d4-ba93b975ec64", "name": "Kafka3ConnectionService", "comments": "", "type": "org.apache.nifi.kafka.service.Kafka3ConnectionService", "bundle": { "group": "org.apache.nifi", "artifact": "nifi-kafka-3-service-nar", "version": "2.6.0" }, "properties": { "security.protocol": "SSL", "SSL Context Service": "7fce889c-fc16-3f69-b7b4-44a90a81ae38", "ack.wait.time": "5 sec", "sasl.username": null, "max.poll.records": "10000", "sasl.mechanism": "GSSAPI", "isolation.level": "read_committed", "max.block.ms": "5 sec", "bootstrap.servers": "kafka-broker-default-headless.default.svc.cluster.local:9093", "kerberos-user-service": null, "sasl.kerberos.service.name": null, "default.api.timeout.ms": "60 sec" }, "propertyDescriptors": { "ack.wait.time": { "name": "ack.wait.time", "displayName": "Acknowledgment Wait Time", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "sasl.username": { "name": "sasl.username", "displayName": "SASL Username", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "max.poll.records": { "name": "max.poll.records", "displayName": "Max Poll Records", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "bootstrap.servers": { "name": "bootstrap.servers", "displayName": "Bootstrap Servers", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "sasl.kerberos.service.name": { "name": "sasl.kerberos.service.name", "displayName": "Kerberos Service Name", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "security.protocol": { "name": "security.protocol", "displayName": "Security Protocol", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "SSL Context Service": { "name": "SSL Context Service", "displayName": "SSL Context Service", "identifiesControllerService": true, "sensitive": false, "dynamic": false }, "sasl.mechanism": { "name": "sasl.mechanism", "displayName": "SASL Mechanism", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "isolation.level": { "name": "isolation.level", "displayName": "Transaction Isolation Level", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "max.block.ms": { "name": "max.block.ms", "displayName": "Max Metadata Wait Time", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "kerberos-user-service": { "name": "kerberos-user-service", "displayName": "Kerberos User Service", "identifiesControllerService": true, "sensitive": false, "dynamic": false }, "sasl.password": { "name": "sasl.password", "displayName": "SASL Password", "identifiesControllerService": false, "sensitive": true, "dynamic": false }, "default.api.timeout.ms": { "name": "default.api.timeout.ms", "displayName": "Client Timeout", "identifiesControllerService": false, "sensitive": false, "dynamic": false } }, "controllerServiceApis": [ { "type": "org.apache.nifi.kafka.service.api.KafkaConnectionService", "bundle": { "group": "org.apache.nifi", "artifact": "nifi-kafka-service-api-nar", "version": "2.6.0" } } ], "scheduledState": "DISABLED", "bulletinLevel": "WARN", "componentType": "CONTROLLER_SERVICE", "groupIdentifier": "a1618dd5-e11a-3ff4-9a6c-62607956b4f4" }, { "identifier": "d082021f-5f72-322a-8e1d-c5990d04b2b1", "instanceIdentifier": "ab119a98-0d96-3556-b54f-411fd9cdd462", "name": "CSVReader", "type": "org.apache.nifi.csv.CSVReader", "bundle": { "group": "org.apache.nifi", "artifact": "nifi-record-serialization-services-nar", "version": "2.6.0" }, "properties": { "schema-reference-reader": null, "ignore-csv-header": "false", "schema-branch": null, "Trim double quote": "true", "CSV Format": "custom", "Quote Character": "\"", "csvutils-allow-duplicate-header-names": "true", "Value Separator": ",", "Record Separator": "\\n", "Timestamp Format": null, "Escape Character": "\\", "Date Format": null, "Null String": null, "Skip Header Line": "false", "Trim Fields": "true", "schema-name": "${schema.name}", "schema-registry": null, "csv-reader-csv-parser": "commons-csv", "Time Format": null, "Comment Marker": null, "schema-access-strategy": "infer-schema", "schema-version": null, "csvutils-character-set": "UTF-8", "schema-text": "${avro.schema}" }, "propertyDescriptors": { "schema-reference-reader": { "name": "schema-reference-reader", "displayName": "Schema Reference Reader", "identifiesControllerService": true, "sensitive": false, "dynamic": false }, "ignore-csv-header": { "name": "ignore-csv-header", "displayName": "Ignore CSV Header Column Names", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "schema-branch": { "name": "schema-branch", "displayName": "Schema Branch", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Trim double quote": { "name": "Trim double quote", "displayName": "Trim double quote", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "CSV Format": { "name": "CSV Format", "displayName": "CSV Format", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Quote Character": { "name": "Quote Character", "displayName": "Quote Character", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "csvutils-allow-duplicate-header-names": { "name": "csvutils-allow-duplicate-header-names", "displayName": "Allow Duplicate Header Names", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Value Separator": { "name": "Value Separator", "displayName": "Value Separator", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Record Separator": { "name": "Record Separator", "displayName": "Record Separator", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Timestamp Format": { "name": "Timestamp Format", "displayName": "Timestamp Format", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Escape Character": { "name": "Escape Character", "displayName": "Escape Character", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Date Format": { "name": "Date Format", "displayName": "Date Format", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Null String": { "name": "Null String", "displayName": "Null String", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Skip Header Line": { "name": "Skip Header Line", "displayName": "Treat First Line as Header", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Trim Fields": { "name": "Trim Fields", "displayName": "Trim Fields", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "schema-name": { "name": "schema-name", "displayName": "Schema Name", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "schema-registry": { "name": "schema-registry", "displayName": "Schema Registry", "identifiesControllerService": true, "sensitive": false, "dynamic": false }, "csv-reader-csv-parser": { "name": "csv-reader-csv-parser", "displayName": "CSV Parser", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Time Format": { "name": "Time Format", "displayName": "Time Format", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Comment Marker": { "name": "Comment Marker", "displayName": "Comment Marker", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "schema-access-strategy": { "name": "schema-access-strategy", "displayName": "Schema Access Strategy", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "schema-version": { "name": "schema-version", "displayName": "Schema Version", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "csvutils-character-set": { "name": "csvutils-character-set", "displayName": "Character Set", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "schema-text": { "name": "schema-text", "displayName": "Schema Text", "identifiesControllerService": false, "sensitive": false, "dynamic": false } }, "controllerServiceApis": [ { "type": "org.apache.nifi.serialization.RecordReaderFactory", "bundle": { "group": "org.apache.nifi", "artifact": "nifi-standard-services-api-nar", "version": "2.6.0" } } ], "scheduledState": "DISABLED", "bulletinLevel": "WARN", "componentType": "CONTROLLER_SERVICE", "groupIdentifier": "a1618dd5-e11a-3ff4-9a6c-62607956b4f4" }, { "identifier": "7fce889c-fc16-3f69-b7b4-44a90a81ae38", "instanceIdentifier": "3238df43-ee04-3bec-aa9b-13efe7c0f72a", "name": "StandardRestrictedSSLContextService", "comments": "", "type": "org.apache.nifi.ssl.StandardRestrictedSSLContextService", "bundle": { "group": "org.apache.nifi", "artifact": "nifi-ssl-context-service-nar", "version": "2.6.0" }, "properties": { "Truststore Type": "PKCS12", "Truststore Password": "secret", "SSL Protocol": "TLSv1.3", "Keystore Type": "PKCS12", "Keystore Password": "secret", "Truststore Filename": "/stackable/keystore/truststore.p12", "Keystore Filename": "/stackable/keystore/keystore.p12" }, "propertyDescriptors": { "Truststore Type": { "name": "Truststore Type", "displayName": "Truststore Type", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "SSL Protocol": { "name": "SSL Protocol", "displayName": "TLS Protocol", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Keystore Type": { "name": "Keystore Type", "displayName": "Keystore Type", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Truststore Filename": { "name": "Truststore Filename", "displayName": "Truststore Filename", "identifiesControllerService": false, "sensitive": false, "dynamic": false, "resourceDefinition": { "cardinality": "SINGLE", "resourceTypes": [ "FILE" ] } }, "Keystore Password": { "name": "Keystore Password", "displayName": "Keystore Password", "identifiesControllerService": false, "sensitive": true, "dynamic": false }, "key-password": { "name": "key-password", "displayName": "Key Password", "identifiesControllerService": false, "sensitive": true, "dynamic": false }, "Truststore Password": { "name": "Truststore Password", "displayName": "Truststore Password", "identifiesControllerService": false, "sensitive": true, "dynamic": false }, "Keystore Filename": { "name": "Keystore Filename", "displayName": "Keystore Filename", "identifiesControllerService": false, "sensitive": false, "dynamic": false, "resourceDefinition": { "cardinality": "SINGLE", "resourceTypes": [ "FILE" ] } } }, "controllerServiceApis": [ { "type": "org.apache.nifi.ssl.SSLContextService", "bundle": { "group": "org.apache.nifi", "artifact": "nifi-standard-services-api-nar", "version": "2.6.0" } }, { "type": "org.apache.nifi.ssl.RestrictedSSLContextService", "bundle": { "group": "org.apache.nifi", "artifact": "nifi-standard-services-api-nar", "version": "2.6.0" } }, { "type": "org.apache.nifi.ssl.SSLContextProvider", "bundle": { "group": "org.apache.nifi", "artifact": "nifi-standard-services-api-nar", "version": "2.6.0" } } ], "scheduledState": "DISABLED", "bulletinLevel": "WARN", "componentType": "CONTROLLER_SERVICE", "groupIdentifier": "a1618dd5-e11a-3ff4-9a6c-62607956b4f4" }, { "identifier": "f3254a5e-0024-31c3-b59a-9edff2a099c9", "instanceIdentifier": "8dd599ec-3c90-3a34-8213-185a6d1bfb92", "name": "JsonRecordSetWriter", "type": "org.apache.nifi.json.JsonRecordSetWriter", "bundle": { "group": "org.apache.nifi", "artifact": "nifi-record-serialization-services-nar", "version": "2.6.0" }, "properties": { "schema-reference-reader": null, "schema-branch": null, "Allow Scientific Notation": "false", "compression-level": "1", "schema-cache": null, "Timestamp Format": null, "Date Format": null, "Pretty Print JSON": "false", "compression-format": "none", "Schema Write Strategy": "no-schema", "suppress-nulls": "never-suppress", "output-grouping": "output-array", "schema-name": "${schema.name}", "schema-registry": null, "Time Format": null, "schema-access-strategy": "inherit-record-schema", "schema-version": null, "schema-text": "${avro.schema}", "Schema Reference Writer": null }, "propertyDescriptors": { "schema-reference-reader": { "name": "schema-reference-reader", "displayName": "Schema Reference Reader", "identifiesControllerService": true, "sensitive": false, "dynamic": false }, "schema-branch": { "name": "schema-branch", "displayName": "Schema Branch", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Allow Scientific Notation": { "name": "Allow Scientific Notation", "displayName": "Allow Scientific Notation", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "compression-level": { "name": "compression-level", "displayName": "Compression Level", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "schema-cache": { "name": "schema-cache", "displayName": "Schema Cache", "identifiesControllerService": true, "sensitive": false, "dynamic": false }, "Timestamp Format": { "name": "Timestamp Format", "displayName": "Timestamp Format", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Date Format": { "name": "Date Format", "displayName": "Date Format", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Pretty Print JSON": { "name": "Pretty Print JSON", "displayName": "Pretty Print JSON", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "compression-format": { "name": "compression-format", "displayName": "Compression Format", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Schema Write Strategy": { "name": "Schema Write Strategy", "displayName": "Schema Write Strategy", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "suppress-nulls": { "name": "suppress-nulls", "displayName": "Suppress Null Values", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "output-grouping": { "name": "output-grouping", "displayName": "Output Grouping", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "schema-name": { "name": "schema-name", "displayName": "Schema Name", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "schema-registry": { "name": "schema-registry", "displayName": "Schema Registry", "identifiesControllerService": true, "sensitive": false, "dynamic": false }, "Time Format": { "name": "Time Format", "displayName": "Time Format", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "schema-access-strategy": { "name": "schema-access-strategy", "displayName": "Schema Access Strategy", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "schema-version": { "name": "schema-version", "displayName": "Schema Version", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "schema-text": { "name": "schema-text", "displayName": "Schema Text", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Schema Reference Writer": { "name": "Schema Reference Writer", "displayName": "Schema Reference Writer", "identifiesControllerService": true, "sensitive": false, "dynamic": false } }, "controllerServiceApis": [ { "type": "org.apache.nifi.serialization.RecordSetWriterFactory", "bundle": { "group": "org.apache.nifi", "artifact": "nifi-standard-services-api-nar", "version": "2.6.0" } } ], "scheduledState": "DISABLED", "bulletinLevel": "WARN", "componentType": "CONTROLLER_SERVICE", "groupIdentifier": "a1618dd5-e11a-3ff4-9a6c-62607956b4f4" }, { "identifier": "5e81a80d-728d-3c87-800e-aa957618da55", "instanceIdentifier": "a53045c6-0195-1000-ffff-ffffef6c13db", "name": "CSVRecordSetWriter", "comments": "", "type": "org.apache.nifi.csv.CSVRecordSetWriter", "bundle": { "group": "org.apache.nifi", "artifact": "nifi-record-serialization-services-nar", "version": "2.6.0" }, "properties": { "schema-reference-reader": null, "schema-branch": null, "CSV Format": "custom", "Include Header Line": "true", "schema-cache": null, "Escape Character": "\\", "Date Format": null, "Null String": null, "Trim Fields": "true", "schema-registry": null, "Time Format": null, "Comment Marker": null, "schema-access-strategy": "inherit-record-schema", "schema-version": null, "csvutils-character-set": "UTF-8", "schema-text": "${avro.schema}", "csv-writer": "commons-csv", "Quote Character": "\"", "Value Separator": ",", "Record Separator": "\\n", "Timestamp Format": null, "Schema Write Strategy": "no-schema", "schema-name": "${schema.name}", "Quote Mode": "MINIMAL", "Schema Reference Writer": null, "Include Trailing Delimiter": "false" }, "propertyDescriptors": { "schema-reference-reader": { "name": "schema-reference-reader", "displayName": "Schema Reference Reader", "identifiesControllerService": true, "sensitive": false, "dynamic": false }, "schema-branch": { "name": "schema-branch", "displayName": "Schema Branch", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "CSV Format": { "name": "CSV Format", "displayName": "CSV Format", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Include Header Line": { "name": "Include Header Line", "displayName": "Include Header Line", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "schema-cache": { "name": "schema-cache", "displayName": "Schema Cache", "identifiesControllerService": true, "sensitive": false, "dynamic": false }, "Escape Character": { "name": "Escape Character", "displayName": "Escape Character", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Date Format": { "name": "Date Format", "displayName": "Date Format", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Null String": { "name": "Null String", "displayName": "Null String", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Trim Fields": { "name": "Trim Fields", "displayName": "Trim Fields", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "schema-registry": { "name": "schema-registry", "displayName": "Schema Registry", "identifiesControllerService": true, "sensitive": false, "dynamic": false }, "Time Format": { "name": "Time Format", "displayName": "Time Format", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Comment Marker": { "name": "Comment Marker", "displayName": "Comment Marker", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "schema-access-strategy": { "name": "schema-access-strategy", "displayName": "Schema Access Strategy", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "schema-version": { "name": "schema-version", "displayName": "Schema Version", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "csvutils-character-set": { "name": "csvutils-character-set", "displayName": "Character Set", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "schema-text": { "name": "schema-text", "displayName": "Schema Text", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "csv-writer": { "name": "csv-writer", "displayName": "CSV Writer", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Quote Character": { "name": "Quote Character", "displayName": "Quote Character", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Value Separator": { "name": "Value Separator", "displayName": "Value Separator", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Record Separator": { "name": "Record Separator", "displayName": "Record Separator", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Timestamp Format": { "name": "Timestamp Format", "displayName": "Timestamp Format", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Schema Write Strategy": { "name": "Schema Write Strategy", "displayName": "Schema Write Strategy", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "schema-name": { "name": "schema-name", "displayName": "Schema Name", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Quote Mode": { "name": "Quote Mode", "displayName": "Quote Mode", "identifiesControllerService": false, "sensitive": false, "dynamic": false }, "Schema Reference Writer": { "name": "Schema Reference Writer", "displayName": "Schema Reference Writer", "identifiesControllerService": true, "sensitive": false, "dynamic": false }, "Include Trailing Delimiter": { "name": "Include Trailing Delimiter", "displayName": "Include Trailing Delimiter", "identifiesControllerService": false, "sensitive": false, "dynamic": false } }, "controllerServiceApis": [ { "type": "org.apache.nifi.serialization.RecordSetWriterFactory", "bundle": { "group": "org.apache.nifi", "artifact": "nifi-standard-services-api-nar", "version": "2.6.0" } } ], "scheduledState": "DISABLED", "bulletinLevel": "WARN", "componentType": "CONTROLLER_SERVICE", "groupIdentifier": "a1618dd5-e11a-3ff4-9a6c-62607956b4f4" } ], "defaultFlowFileExpiration": "0 sec", "defaultBackPressureObjectThreshold": 10000, "defaultBackPressureDataSizeThreshold": "1 GB", "scheduledState": "ENABLED", "executionEngine": "INHERITED", "maxConcurrentTasks": 1, "statelessFlowTimeout": "1 min", "flowFileConcurrency": "UNBOUNDED", "flowFileOutboundPolicy": "STREAM_WHEN_AVAILABLE", "componentType": "PROCESS_GROUP" }, "externalControllerServices": {}, "parameterContexts": {}, "flowEncodingVersion": "1.0", "parameterProviders": {}, "latest": false } diff --git a/docs/modules/demos/pages/nifi-kafka-druid-earthquake-data.adoc b/docs/modules/demos/pages/nifi-kafka-druid-earthquake-data.adoc index 0dce404f..feb95649 100644 --- a/docs/modules/demos/pages/nifi-kafka-druid-earthquake-data.adoc +++ b/docs/modules/demos/pages/nifi-kafka-druid-earthquake-data.adoc @@ -27,7 +27,7 @@ Additionally, we have to use the FQDN service names (including the namespace), s To run this demo, your system needs at least: * 9 {k8s-cpu}[cpu units] (core/hyperthread) -* 42GiB memory +* 42GiB memory (minimum of 16GiB per node) * 75GiB disk storage == Overview @@ -86,6 +86,111 @@ $ stackablectl stacklet list include::partial$instance-hint.adoc[] +== Inspect the data in Kafka + +Kafka is an event streaming platform to stream the data in near real-time. +All the messages put in and read from Kafka are structured in dedicated queues called topics. +The test data will be put into a topic called `earthquakes`. +The records are produced (written) by the test data generator and consumed (read) by Druid afterwards in the same order they were created. + +Kafka uses mutual TLS, so clients wanting to connect to Kafka must present a valid TLS certificate. +The easiest way to obtain this is to shell into the `kafka-broker-default-0` Pod, as we will do in the following section for demonstration purposes. +For a production setup, you should spin up a dedicated Pod provisioned with a certificate acting as a Kafka client instead of shell-ing into the Kafka Pod. + +=== List the available Topics + +You can execute a command on the Kafka broker to list the available topics as follows: + +// In the following commands the kcat-prober container instead of the kafka container is used to send requests to Kafka. +// This is necessary because kcat cannot use key- and truststore files with empty passwords, which are mounted here to the kafka container. +// However, the kcat-prober container has TLS certificates mounted, which can be used by kcat to connect to Kafka. +[source,console] +---- +$ kubectl exec kafka-broker-default-0 -c kafka -- \ +/stackable/kafka/bin/kafka-topics.sh \ +--describe \ +--bootstrap-server kafka-broker-default-headless.default.svc.cluster.local:9093 \ +--command-config /stackable/config/client.properties +... +Topic: earthquakes TopicId: ND51v_XcQPK4Ilm7A35Pag PartitionCount: 8 ReplicationFactor: 1 Configs: min.insync.replicas=1,segment.bytes=100000000,retention.bytes=900000000 + Topic: earthquakes Partition: 0 Leader: 1243966388 Replicas: 1243966388 Isr: 1243966388 Elr: LastKnownElr: + Topic: earthquakes Partition: 1 Leader: 1243966388 Replicas: 1243966388 Isr: 1243966388 Elr: LastKnownElr: + Topic: earthquakes Partition: 2 Leader: 1243966388 Replicas: 1243966388 Isr: 1243966388 Elr: LastKnownElr: + Topic: earthquakes Partition: 3 Leader: 1243966388 Replicas: 1243966388 Isr: 1243966388 Elr: LastKnownElr: + Topic: earthquakes Partition: 4 Leader: 1243966388 Replicas: 1243966388 Isr: 1243966388 Elr: LastKnownElr: + Topic: earthquakes Partition: 5 Leader: 1243966388 Replicas: 1243966388 Isr: 1243966388 Elr: LastKnownElr: + Topic: earthquakes Partition: 6 Leader: 1243966388 Replicas: 1243966388 Isr: 1243966388 Elr: LastKnownElr: + Topic: earthquakes Partition: 7 Leader: 1243966388 Replicas: 1243966388 Isr: 1243966388 Elr: LastKnownElr: +---- + +You can see that Kafka consists of one broker, and the topic `earthquakes` with eight partitions has been created. To +see some records sent to Kafka, run the following command. You can change the number of records to print via the `--max-messages` +parameter. + +[source,console] +---- +$ kubectl exec kafka-broker-default-0 -c kafka -- \ +/stackable/kafka/bin/kafka-console-consumer.sh \ +--bootstrap-server kafka-broker-default-headless.default.svc.cluster.local:9093 \ +--consumer.config /stackable/config/client.properties \ +--topic earthquakes \ +--offset earliest \ +--partition 0 \ +--max-messages 1 +---- + +Below is an example of the output of one record: + +[source,json] +---- + { + "time":"1950-02-07T10:37:29.240Z", + "latitude":45.949, + "longitude":151.59, + "depth":35.0, + "mag":5.94, + "magType":"mw", + "nst":null, + "gap":null, + "dmin":null, + "rms":null, + "net":"iscgem", + "id":"iscgem895202", + "updated":"2022-04-26T18:23:38.377Z", + "place":"Kuril Islands", + "type":"earthquake", + "horizontalError":null, + "depthError":12.6, + "magError":0.55, + "magNst":null, + "status":"reviewed", + "locationSource":"iscgem", + "magSource":"iscgem" +} +---- + +If you are interested in how many records have been produced to the Kafka topic so far, use the following command. + +[source,console] +---- +$ kubectl exec kafka-broker-default-0 -c kafka -- \ +/stackable/kafka/bin/kafka-get-offsets.sh \ +--bootstrap-server kafka-broker-default-headless.default.svc.cluster.local:9093 \ +--command-config /stackable/config/client.properties \ +--topic earthquakes +... +earthquakes:0:757379 +earthquakes:1:759282 +earthquakes:2:761924 +earthquakes:3:761339 +earthquakes:4:759059 +earthquakes:5:767695 +earthquakes:6:771457 +earthquakes:7:768301 +---- + +If you calculate `765,000` records * `8` partitions, you end up with ~ 6,120,000 records. + == NiFi NiFi is used to fetch earthquake data from the internet and ingest it into Kafka. diff --git a/stacks/nifi-kafka-druid-superset-s3/nifi.yaml b/stacks/nifi-kafka-druid-superset-s3/nifi.yaml index bb46f7ad..f7e3fcdf 100644 --- a/stacks/nifi-kafka-druid-superset-s3/nifi.yaml +++ b/stacks/nifi-kafka-druid-superset-s3/nifi.yaml @@ -21,7 +21,7 @@ spec: min: "500m" max: "4" memory: - limit: '6Gi' + limit: "6Gi" storage: contentRepo: capacity: "10Gi" @@ -33,6 +33,10 @@ spec: capacity: "4Gi" stateRepo: capacity: "1Gi" + configOverrides: + nifi.properties: + nifi.web.https.sni.required: "false" + nifi.web.https.sni.host.check: "false" roleGroups: default: replicas: 1