Python脚本如何生成NiFi流程配置

wen 实用脚本 2026-07-04 23

本文目录导读：

方式一：直接操作 Flow.xml.gz 文件
方式二：使用 NiFi API 封装工具
方式三：直接生成 JSON 格式流程配置
方式四：使用 NiFi Registry API
最佳实践建议

在 Apache NiFi 中，生成流程配置通常是指生成 Flow.xml.gz (或 Flow.json.gz) 文件，你可以通过编写 Python 脚本来直接生成或修改这个文件。

以下提供几种主要的实现方式：

直接操作 Flow.xml.gz 文件

NiFi 在 ./conf/flow.xml.gz 中存储流程配置，这是一个压缩的 XML 文件，最直接的方式就是解析和修改这个文件。

import gzip
import xml.etree.ElementTree as ET
def parse_flow_xml():
    """读取并解析现有的 flow.xml.gz 文件"""
    with gzip.open('conf/flow.xml.gz', 'rb') as f:
        tree = ET.parse(f)
        root = tree.getroot()
    return tree, root
def save_flow_xml(tree):
    """保存修改后的 flow.xml.gz 文件"""
    with gzip.open('conf/flow.xml.gz', 'wb') as f:
        tree.write(f, encoding='utf-8', xml_declaration=True)
def add_processor(root, processor_id, name, type_class, position_x=0, position_y=0):
    """添加一个处理器到流程中"""
    processor_group = root.find('.//processorGroup')
    # 创建处理器元素
    processor = ET.SubElement(processor_group, 'processor')
    processor.set('id', processor_id)
    # 添加基本信息
    name_elem = ET.SubElement(processor, 'name')
    name_elem.text = name
    type_elem = ET.SubElement(processor, 'type')
    type_elem.text = type_class
    # 添加位置信息
    position = ET.SubElement(processor, 'position')
    position.set('x', str(position_x))
    position.set('y', str(position_y))
    return processor
def create_simple_flow():
    """创建一个简单的示例流程"""
    tree, root = parse_flow_xml()
    # 添加 GenerateFlowFile 处理器
    add_processor(root, 'processor-1', 'Generate Data', 
                  'org.apache.nifi.processors.standard.GenerateFlowFile',
                  position_x=100, position_y=200)
    # 添加 LogAttribute 处理器
    add_processor(root, 'processor-2', 'Log Data',
                  'org.apache.nifi.processors.standard.LogAttribute',
                  position_x=400, position_y=200)
    # 创建连接
    connection = ET.SubElement(root.find('.//connections'), 'connection')
    connection.set('id', 'connection-1')
    source = ET.SubElement(connection, 'source')
    source.set('id', 'processor-1')
    source.set('type', 'PROCESSOR')
    destination = ET.SubElement(connection, 'destination')
    destination.set('id', 'processor-2')
    destination.set('type', 'PROCESSOR')
    save_flow_xml(tree)
    print("流程配置已生成")

使用 NiFi API 封装工具

使用第三方库可以更方便地操作 NiFi 流程：

1 使用 nipyapi 库

pip install nipyapi

import nipyapi
from nipyapi import canvas
# 连接到 NiFi 实例
nipyapi.config.nifi_config.host = 'http://localhost:8080'
def create_flow_via_api():
    """通过 API 创建流程"""
    # 获取根进程组
    root_pg = canvas.get_root_pg_id()
    # 创建处理器
    generate_flowfile = canvas.create_processor(
        parent_pg=root_pg,
        processor='GenerateFlowFile',
        name='Generate Data',
        location=(100, 200)
    )
    log_attribute = canvas.create_processor(
        parent_pg=root_pg,
        processor='LogAttribute',
        name='Log Data',
        location=(400, 200)
    )
    # 创建连接
    connection = canvas.create_connection(
        parent_pg=root_pg,
        source=generate_flowfile,
        target=log_attribute
    )
    print("流程已创建")
def create_complex_flow():
    """创建更复杂的流程，包含配置"""
    root_pg = canvas.get_root_pg_id()
    # 创建带有配置的处理器
    update_attribute = canvas.create_processor(
        parent_pg=root_pg,
        processor='UpdateAttribute',
        name='Update Attributes',
        location=(250, 200),
        config={
            'properties': {
                'my-custom-attribute': '${filename}',
                'timestamp': '${now()}'
            }
        }
    )

2 使用 niiflowgen 库

专门用于生成 NiFi 流程配置的库：

pip install niiflowgen

from niiflowgen import FlowBuilder, FlowElement
def generate_flow_config():
    """生成流程配置文件"""
    builder = FlowBuilder()
    # 创建处理器
    processor1 = FlowElement(
        type='GenerateFlowFile',
        name='Generate Data',
        position={'x': 100, 'y': 200},
        properties={
            'file.size': '1 KB'
        }
    )
    # 添加处理器到流程
    builder.add_processor('generate', processor1)
    # 创建第二个处理器
    processor2 = FlowElement(
        type='LogAttribute',
        name='Log Data',
        position={'x': 400, 'y': 200}
    )
    builder.add_processor('log', processor2)
    # 添加连接
    builder.add_connection('generate', 'log')
    # 生成流配置文件
    flow_config = builder.build()
    # 保存配置文件
    with open('generated_flow.json', 'w') as f:
        json.dump(flow_config, f, indent=2)
    print("流程配置已生成")

直接生成 JSON 格式流程配置

NiFi 1.x 版本支持 JSON 格式的流程导出和导入：

import json
from datetime import datetime
def create_nifi_flow_json():
    """创建 NiFi 流程 JSON 配置"""
    flow_config = {
        "flowContents": {
            "identifier": "root-process-group",
            "name": "My Flow",
            "comments": "",
            "position": {"x": 0, "y": 0},
            "processGroups": [],
            "processors": [
                {
                    "identifier": "generate-flowfile",
                    "name": "GenerateData",
                    "type": "org.apache.nifi.processors.standard.GenerateFlowFile",
                    "position": {"x": 100, "y": 200},
                    "config": {
                        "properties": {
                            "File Size": "1 KB",
                            "Batch Size": "1"
                        },
                        "schedulingPeriod": "0 sec",
                        "schedulingStrategy": "TIMER_DRIVEN"
                    }
                },
                {
                    "identifier": "log-attribute",
                    "name": "LogData",
                    "type": "org.apache.nifi.processors.standard.LogAttribute",
                    "position": {"x": 400, "y": 200},
                    "config": {
                        "properties": {
                            "Log Level": "info",
                            "Attributes to Log": ".*"
                        }
                    }
                }
            ],
            "connections": [
                {
                    "identifier": "conn-1",
                    "name": "",
                    "source": {
                        "id": "generate-flowfile",
                        "type": "PROCESSOR"
                    },
                    "destination": {
                        "id": "log-attribute",
                        "type": "PROCESSOR"
                    },
                    "selectedRelationships": ["success"],
                    "backPressureObjectThreshold": 10000,
                    "backPressureDataSizeThreshold": "1 GB"
                }
            ],
            "controllerServices": []
        },
        "templateTags": ["generated", "python"],
        "templateName": "Generated Flow {}".format(datetime.now().strftime("%Y%m%d_%H%M%S"))
    }
    # 保存为 JSON 文件（可用于 NiFi 模板导入）
    with open('nifi_flow_template.json', 'w') as f:
        json.dump(flow_config, f, indent=2)
    print("JSON 流程配置已生成")
    # 如果需要生成 flow.xml.gz 格式
    import gzip
    import xml.etree.ElementTree as ET
    # 转换为 XML 格式（简化版）
    root = ET.Element("flowController")
    # ... 添加 XML 结构 ...
    with gzip.open('flow.xml.gz', 'wt', encoding='utf-8') as f:
        f.write(ET.tostring(root, encoding='unicode'))

使用 NiFi Registry API

如果你的环境有 NiFi Registry，可以通过 API 创建版本控制流程：

import requests
import json
def create_flow_via_registry():
    """通过 Registry API 创建流程"""
    base_url = "http://localhost:18080/nifi-registry-api"
    # 创建 Bucket
    bucket_data = {
        "name": "My Flows",
        "description": "Generated flows"
    }
    response = requests.post(f"{base_url}/buckets", 
                            json=bucket_data)
    bucket_id = response.json()["identifier"]
    # 创建流程版本
    flow_data = {
        "name": "Generated Flow",
        "description": "Flow generated by Python script",
        "bucketIdentifier": bucket_id,
        "flowContents": {
            # 此处使用上面生成的 flow_config 内容
        }
    }
    response = requests.post(f"{base_url}/buckets/{bucket_id}/flows",
                            json=flow_data)
    print(f"Flow created: {response.json()}")

最佳实践建议

使用模板方式：更推荐使用 NiFi 模板（JSON/XML），通过 API 导入
测试流程：生成后在测试环境验证
版本控制：将生成的流程配置纳入 Git 管理
错误处理：添加适当的异常处理和日志记录
使用 NiFi CLI：结合 nifi-toolkit 的命令行工具

选择哪种方式取决于你的具体需求：

如果需要修改现有流程：方式一
如果开发新流程且使用 NiFi 2.x：方式二
如果需要生成模板文件：方式三