Python Evaluator*¶

支持运行原生的 Python 脚本，实现自定义的业务逻辑。与旧数据格式不兼容，即无法按 ModelId::PointId 进行数据过滤。

配置详情¶

该算子的配置包括 General，Basic，Input/Output，和 Script 的详细信息，各字段的配置如下：

General¶

名称	是否必须	描述
Name	Yes	算子名称
Description	No	算子描述
Stage Library	Yes	算子所属的库
Required Fields	No	数据必须包含的字段，如果未包含指定字段，则 record 将被过滤掉
Preconditions	No	数据必须满足的前提条件，如果不满足指定条件，则 record 将被过滤掉。例如：`${record:value('/value') > 0}`。有关 EL 语句的使用方法，参考 Expression Language。
On Record Error	Yes	对错误数据的处理方式，可选： Discard：直接丢弃 Send to Error：发送至错误中心 Stop Pipeline：停止流任务运行

Basic¶

名称

是否必须

描述

Lineage Mapping

Yes

选择数据输入点和数据输出点的血缘对应关系。可选：

1:1：Input/Output 参数配置中，一个数据输入点对应一个数据输出点
M:1：Input/Output 参数配置中，可配置任意数量的数据输入点和数据输出点

Quality Filter

No

根据数据质量过滤处理数据，只有符合质量条件的 record 才会进行此次处理

Input/Output¶

名称	是否必须	描述
Input Measurement	Yes	数据输入点，输入数据的 MeasurementId 必须匹配输入点，才能够进入后续计算。
Output Measurement	No	数据输出点，经过 Python Script 脚本后的输出数据的 MeasurementId 必须匹配输出点，才能够作为真正的输出 record。

Script¶

名称	是否必须	描述
Python Script	Yes	编写自定义 Python 脚本。其中 records 代表所有经过选中的点并经过质量控制后，流入的数据。

输出结果¶

运行自定义 Python 脚本后，该算子的输出结果包含在 attr 结构体中。

输出示例¶

../../../_images/python_evaluator_result11.png

Python 脚本开发指南¶

# Available constants:

   They are to assign a type to a field with a value null.

   NULL_BOOLEAN, NULL_CHAR, NULL_BYTE, NULL_SHORT, NULL_INTEGER, NULL_LONG

   NULL_FLOATNULL_DOUBLE, NULL_DATE, NULL_DATETIME, NULL_TIME, NULL_DECIMAL

   NULL_BYTE_ARRAY, NULL_STRING, NULL_LIST, NULL_MAP

# Available Objects:
records: an array of records to process, depending on Jython processor processing mode it may have 1 record or all the records in the batch.

state: a dict that is preserved between invocations of this script.  Useful for caching bits of data e.g. counters.

log.<loglevel>(msg, obj...):
use instead of print to send log messages to the log4j log instead of stdout.
loglevel is any log4j level: e.g. info, error, warn, trace.

output.write(record): writes a record to processor output

error.write(record, message): sends a record to error

sdcFunctions.getFieldNull(Record, 'field path'): Receive a constant defined above to check if the field is typed field with value null

sdcFunctions.createRecord(String recordId): Creates a new record.

Pass a recordId to uniquely identify the record and include enough information to track down the record source.

sdcFunctions.createMap(boolean listMap): Create a map for use as a field in a record.
Pass True to this function to create a list map (ordered map)

sdcFunctions.createEvent(String type, int version): Creates a new event.
Create new empty event with standard headers.

sdcFunctions.toEvent(Record): Send event to event stream
Only events created with sdcFunctions.createEvent are supported.

sdcFunctions.isPreview(): Determine if pipeline is in preview mode.

Available Record Header Variables:

record.attributes: a map of record header attributes.

record.<header name>: get the value of 'header name'.

# Add additional module search paths:

import sys
sys.path.append('/some/other/dir/to/search')

for record in records:

  try:

    # write record to processor output
    output.write(record)
  except Exception as e:
    # trace the exception
    import sys
    error.trace(sys.exc_info())
    # Send record to error
    error.write(record, str(e))