Apache Nifi使用多个输入的EvaluateJsonPath
创始人
2024-09-04 13:00:50
0

在Apache Nifi中,您可以使用EvaluateJsonPath处理器来从JSON数据中提取特定字段的值。如果您需要处理多个输入,您可以使用MergeContent处理器将它们合并到一个流中,然后将该流作为EvaluateJsonPath的输入。

以下是一个使用多个输入的EvaluateJsonPath的示例流程:

  1. 在流程中添加一个MergeContent处理器,将多个输入流合并为一个流。配置MergeContent处理器的属性如下:

    • Merge Strategy:使用"Defragment"策略
    • Minimum Number of Entries:设置合并的最小条目数
  2. 在MergeContent处理器后添加一个EvaluateJsonPath处理器。配置EvaluateJsonPath处理器的属性如下:

    • JsonPath Expression:指定要提取的字段的JSON路径表达式
    • Destination:指定提取的字段的属性名
  3. 在EvaluateJsonPath处理器后添加一个LogAttribute处理器,用于将提取的字段值输出到日志中。

以下是一个示例的Groovy脚本,可以使用Apache Nifi的REST API将上述流程导入到Nifi中:

import groovyx.net.http.RESTClient

def baseUrl = 'http://localhost:8080/nifi-api'
def clientId = 'your-client-id'
def processGroupId = 'your-process-group-id'

def restClient = new RESTClient(baseUrl)

// 创建一个新的流程
def createProcessResponse = restClient.post(
    path: '/process-groups/' + processGroupId + '/process-groups',
    headers: [
        'Content-Type': 'application/json',
        'Accept': 'application/json'
    ],
    body: [
        'revision': [
            'clientId': clientId,
            'version': 0
        ],
        'component': [
            'name': 'Multi-input EvaluateJsonPath',
            'position': [
                'x': 0,
                'y': 0
            ]
        ]
    ]
)

def processGroup = createProcessResponse.data
def processGroupId2 = processGroup.id

// 添加MergeContent处理器
def mergeContentProcessor = [
    'revision': [
        'clientId': clientId,
        'version': 0
    ],
    'component': [
        'name': 'MergeContent',
        'type': 'org.apache.nifi.processors.standard.MergeContent',
        'position': [
            'x': 200,
            'y': 200
        ],
        'bundle': [
            'group': 'org.apache.nifi',
            'artifact': 'nifi-standard-nar',
            'version': 'x.y.z'
        ],
        'config': {
            'mergeStrategy': 'Defragment',
            'minimumNumberOfEntries': '2'
        }
    ]
]

def addMergeContentResponse = restClient.post(
    path: '/process-groups/' + processGroupId2 + '/processors',
    headers: [
        'Content-Type': 'application/json',
        'Accept': 'application/json'
    ],
    body: mergeContentProcessor
)

def mergeContentProcessorId = addMergeContentResponse.data.id

// 添加EvaluateJsonPath处理器
def evaluateJsonPathProcessor = [
    'revision': [
        'clientId': clientId,
        'version': 0
    ],
    'component': [
        'name': 'EvaluateJsonPath',
        'type': 'org.apache.nifi.processors.standard.EvaluateJsonPath',
        'position': [
            'x': 400,
            'y': 200
        ],
        'bundle': [
            'group': 'org.apache.nifi',
            'artifact': 'nifi-standard-nar',
            'version': 'x.y.z'
        ],
        'config': {
            'jsonPathExpression': '$.field',
            'destination': 'field'
        }
    ]
]

def addEvaluateJsonPathResponse = restClient.post(
    path: '/process-groups/' + processGroupId2 + '/processors',
    headers: [
        'Content-Type': 'application/json',
        'Accept': 'application/json'
    ],
    body: evaluateJsonPathProcessor
)

def evaluateJsonPathProcessorId = addEvaluateJsonPathResponse.data.id

// 添加LogAttribute处理器
def logAttributeProcessor = [
    'revision': [
        'clientId': clientId,
        'version': 0
    ],
    'component': [
        'name': 'LogAttribute',
        'type': 'org.apache.nifi.processors.attributes.LogAttribute',
        'position': [
            'x': 600,
            'y': 200
        ],
        'bundle': [
            'group': 'org.apache

相关内容

热门资讯

微扑克辅助机器人!微扑克ai机... 微扑克辅助机器人!微扑克ai机器人,wpk微扑克真的有辅助插件(其实真的有挂);无聊就玩这款微扑克真...
aapoker透明挂!wepo... aapoker透明挂!wepoke辅助器真的有用,wopoker辅助v3.0,2025版教程(有挂详...
今日百科!369麻将有挂吗(辅... 今日百科!369麻将有挂吗(辅助)原来存在有挂(2020已更新)(哔哩哔哩)1、下载好369麻将有挂...
7分钟辅助挂!微信小程序边锋干... 7分钟辅助挂!微信小程序边锋干瞪眼辅助器,pokermastersteam本来有挂,2025新版教程...
微扑克wpk透视辅助!微扑克w... 微扑克wpk透视辅助!微扑克wpk辅助软件(辅助挂)切实真的是有挂是一款可以让一直输的玩家,快速成为...
wepower有外 挂!wep... wepower有外 挂!wepoke线上,Wepoke一直真的有挂,揭秘教程(有挂辅助);玩家必备必...
玩家必看!欢乐驿站可以开挂吗(... 玩家必看!欢乐驿站可以开挂吗(透明挂)一贯存在有挂(2026已更新)(哔哩哔哩)1、起透看视 欢乐驿...
二分钟辅助!边锋老友麻将为什么... 二分钟辅助!边锋老友麻将为什么经常输,德州一贯真的是有挂,爆料教程(有挂详情);1、这是跨平台的边锋...
微扑克系统发牌规律!微扑克有辅... 微扑克系统发牌规律!微扑克有辅助,微扑克插件,新2025教程(有挂总结);致您一封信;亲爱微扑克玩家...
wepoke有挂!wepoke... 1、wepoke有挂!wepokeapp下载教程(透明挂)we poker(好像真的有挂);详细教程...