1.新增处理器processor: GenerateFlowFile并增加参数属性

GenerateFlowFile-->properties-->点右上角的+号Add Property-->输入要传入的参数

本例参数为：account and password.

被请求服务器端给出帐号与密码，主要是目的是获取Token，再用Token取查询获取具体的数据。

2.参数属性转换为JSON：AttributesToJSON

 AttributesToJSON-->properties-->Attributes List参数属性输入此处：account,password，此处理器将参数属性转为JSON

-->Destination转换好的json输出为文件内容：flowfile-content。

3.免密获取Token调用API：ExecuteScript

3.1.ExecuteScript-->properties-->Property:Script Engine，Value:Groovy

3.2.ExecuteScript-->properties-->Property:Script Body，Value具体代码如下：

import org.apache.commons.io.IOUtils
import java.nio.charset.StandardCharsets
import org.apache.nifi.processor.io.StreamCallback
//Refer to URL:https://nifi.apache.org/docs/nifi-docs/components/org.apache.nifi/nifi-scripting-nar/1.9.2/org.apache.nifi.processors.script.ExecuteScript/additionalDetails.html

// 1. 定义忽略证书验证的逻辑
def disableSSLVerification = {
javax.net.ssl.HttpsURLConnection.setDefaultHostnameVerifier { hostname, session -> true }
def trustAllCerts = [ new javax.net.ssl.X509TrustManager() {
public void checkClientTrusted(java.security.cert.X509Certificate[] certs, String authType) {}
public void checkServerTrusted(java.security.cert.X509Certificate[] certs, String authType) {}
public java.security.cert.X509Certificate[] getAcceptedIssuers() { null }
} ] as javax.net.ssl.TrustManager[]
def sslContext = javax.net.ssl.SSLContext.getInstance("TLS")
sslContext.init(null, trustAllCerts, new java.security.SecureRandom())
javax.net.ssl.HttpsURLConnection.setDefaultSSLSocketFactory(sslContext.socketFactory)
}

flowFile = session.get()
if(!flowFile) return

try{
disableSSLVerification() // 禁用 SSL 验证

//2.发送 HTTP 请求.Setting request charset ... parameter.
def url = flowFile.getAttribute('Request_token_URL').toURL()
def connection = url.openConnection() as HttpURLConnection
connection.setRequestMethod("POST")
connection.setRequestProperty("Content-Type", "application/json; charset=UTF-8")
connection.setRequestProperty("Accept-Charset", "UTF-8")
connection.setDoOutput(true)

//3.准备Request Body(示例：直接使用字符串，或从 FlowFile 读取).
//Get parameter data:account and password for JSON.
def String read_data = ""
session.read(flowFile, {inputStream ->
read_data = IOUtils.toString(inputStream, StandardCharsets.UTF_8);
} as InputStreamCallback)
def requestBody = read_data // 或从 flowFile 读取
flowFile = session.putAttribute(flowFile, 'requestBody', requestBody)

//4.写入 Body
def outputStream = connection.getOutputStream()
outputStream.write(requestBody.getBytes(StandardCharsets.UTF_8))
outputStream.close()

//5.处理响应并写入 FlowFile.Get response body setting charset to UTF-8.
def responseCode = connection.getResponseCode()
def responseBody = connection.getInputStream().getText("UTF-8")

// Cast a closure with an outputStream parameter to OutputStreamCallback
flowFile = session.write(flowFile, {out ->
out.write(responseBody.getBytes(StandardCharsets.UTF_8))
} as OutputStreamCallback)

session.transfer(flowFile, REL_SUCCESS)
} catch (Exception e) {
log.error("Fails to invoke API for abnormal.", e)
session.transfer(flowFile, REL_FAILURE)
}

3.3.Request Body参数请求体通过上面的json文件传入。

4.解析JSON获取Token：EvaluateJsonPath

EvaluateJsonPath-->properties-->Destination设置输出为属性：flowfile-attribute

-->点右上角的+号Add Property-->Property：输入属性变量token(自己定义),value：$.response.token

 5.设置实际请求参数：UpdateAttribute

UpdateAttribute-->properties-->点右上角的+号Add Property-->输入要传入的参数属性：Property：START_TIME，Value:2023-12-01 00:00:00

Property：END_TIME，Value:2023-12-15 23:59:59

Property：Bearer_Token，Value:Bearer ${token}

6.实际请求参数属性转换为JSON：AttributesToJSON

 AttributesToJSON-->properties-->Attributes List参数属性输入此处：START_TIME,END_TIME，此处理器将参数属性转为JSON

-->Destination转换好的json输出为文件内容：flowfile-content。

7.免密获取数据调用API：ExecuteScript

7.1.ExecuteScript-->properties-->Property:Script Engine，Value:Groovy

7.2.ExecuteScript-->properties-->Property:Script Body，Value具体代码如下：

import org.apache.commons.io.IOUtils
import java.nio.charset.StandardCharsets
import org.apache.nifi.processor.io.StreamCallback
//Refer to URL:https://nifi.apache.org/docs/nifi-docs/components/org.apache.nifi/nifi-scripting-nar/1.9.2/org.apache.nifi.processors.script.ExecuteScript/additionalDetails.html

// 1. 定义忽略证书验证的逻辑
def disableSSLVerification = {
javax.net.ssl.HttpsURLConnection.setDefaultHostnameVerifier { hostname, session -> true }
def trustAllCerts = [ new javax.net.ssl.X509TrustManager() {
public void checkClientTrusted(java.security.cert.X509Certificate[] certs, String authType) {}
public void checkServerTrusted(java.security.cert.X509Certificate[] certs, String authType) {}
public java.security.cert.X509Certificate[] getAcceptedIssuers() { null }
} ] as javax.net.ssl.TrustManager[]
def sslContext = javax.net.ssl.SSLContext.getInstance("TLS")
sslContext.init(null, trustAllCerts, new java.security.SecureRandom())
javax.net.ssl.HttpsURLConnection.setDefaultSSLSocketFactory(sslContext.socketFactory)
}

flowFile = session.get()
if(!flowFile) return

try{
disableSSLVerification() // 禁用 SSL 验证

//2.发送 HTTP 请求.设置请求token、字符集等参数.
//2.1.The URL from GenerateFlowFile processor,by attribute setting.
def url = flowFile.getAttribute('Request_Abnormal_URL').toURL()
def connection = url.openConnection() as HttpURLConnection
connection.setRequestMethod("POST")
connection.setRequestProperty("Content-Type", "application/json; charset=UTF-8")
connection.setRequestProperty("Accept-Charset", "UTF-8") //Charset setting to UTF-8.
//2.2.By token parameter:Bearer + token， request from HTTPS.Setting token parameter.
connection.setRequestProperty("Authorization", flowFile.getAttribute('Bearer_Token'))
connection.setDoOutput(true)

//3.准备Request Body(示例：直接使用字符串，或从 FlowFile 读取).
//Get parameter data:account and password for JSON.
def String read_data = ""
session.read(flowFile, {inputStream ->
read_data = IOUtils.toString(inputStream, StandardCharsets.UTF_8);
} as InputStreamCallback)
def requestBody = read_data // 或从 flowFile 读取
flowFile = session.putAttribute(flowFile, 'requestBody', requestBody)

//4.写入 Setting Request Body.
def outputStream = connection.getOutputStream()
outputStream.write(requestBody.getBytes(StandardCharsets.UTF_8))
outputStream.close()

//5.处理响应并写入 FlowFile.Get response body setting charset to UTF-8.
def responseCode = connection.getResponseCode()
def responseBody = connection.getInputStream().getText("UTF-8")

// Cast a closure with an outputStream parameter to OutputStreamCallback
flowFile = session.write(flowFile, {out ->
out.write(responseBody.getBytes(StandardCharsets.UTF_8))
} as OutputStreamCallback)

session.transfer(flowFile, REL_SUCCESS)
} catch (Exception e) {
log.error("Fails to invoke API for abnormal.", e)
session.transfer(flowFile, REL_FAILURE)
}

7.3.Request Body参数请求体通过上面的json文件传入。