Accessing Data and Metadata

The commons configuration for Connect File Pulse.

Some filters (e.g : AppendFilter can be configured using Simple Connect Expression Language.

Simple Connect Expression Language (ScEL for short) is an expression language based on regex that allows quick access and manipulating record fields and metadata.

The syntaxes to define an expression are of the form : <expression string> or "{{ <expression string> }}".

ScEL supports the following capabilities :

  • Literal expressions
  • Field Selector
  • Nested Navigation
  • String substitution
  • Functions

Literal expressions

  • String : 'Hello World'
  • Number : 42
  • Boolean: True
  • Nullable: null

Field Selector

The expression language can be used to easily select one field from the input record :

$.username

Nested Navigation

To navigate down a struct value, just use a period to indicate a nested field value :

$.address.city

String substitution

The expression language can be used to easily build a new string field that concatenate multiple ones :

The user {{ $.username }} is living in city {{ $.address.city }}

Function

The expression language support function call :

The user {{ $.username }} is living in city {{ uppercase($.address.city) }}

Dynamic Field Selector

String substitution can be used to dynamically select a field :

The bellow example shows how to dynamically build a field selector by concatenating $. and the first element present in the array field $.values.

{{ '$.'extract_array($.values, 0) }}

Note the used of double-quotes to define a substitution expressions

Built-in Functions

ScEL supports a number of predefined functions that can be used to apply a single transformation on a field.

FunctionDescriptionSyntax
concatConcatenate two or more string expressions.{{ concat(expr1, expr2, ...) }}
concat_wsConcatenate two or more string expressions, using the specified separator between each.{{ concat_ws(separator, prefix, suffix, expr1, expr2, ...) }}
containsReturns true if an array field’s value contains the specified value{{ contains(array, 'value') }}
convertsConverts a field’s value into the specified type{{ converts(field, INTEGER) }}
ends_withReturns true if an a string field’s value end with the specified string suffix{{ ends_with(field, 'suffix') }}
equalsReturns true if an a string or number fields’s value equals the specified value{{ equals(field, value) }}
existsReturns true if an the specified field exists{{ ends_with(field, value) }}
extract_arrayReturns the element at the specified position of the specified array{{ extract_array(array, 0) }}
hashHash a given string expression, using murmur2 algorithm{{ hash(field_expr) }}
is_nullReturns true if a field’s value is null{{ is_null(field) }}
lengthReturns the number of elements into an array of the length of an string field{{ length(array) }}
lowercaseConverts all of the characters in a string field’s value to lower case{{ lowercase(field) }}
matchesReturns true if a field’s value match the specified regex{{ matches(field, 'regex') }}
md5Compute the MD5 hash of string expression{{ md5(field_expr) }}
nlvSets a default value if a field’s value is null{{ length(array) }}
replace_allReplaces every subsequence of the field’s value that matches the given pattern with the given replacement string.{{ replace_all(field, 'regex', 'replacement') }}
starts_withReturns true if an a string field’s value start with the specified string prefix{{ starts_with(field, 'prefix') }}
trimTrims the spaces from the beginning and end of a string.{{ trim(field) }}
uppercaseConverts all of the characters in a string field’s value to upper case{{ uppercase(field) }}
uuidCreate a Universally Unique Identifier (UUID){{ uuid() }}

In addition, ScEL supports nested functions.

For example, the following expression can be used to replace all whitespace characters after transforming our field’s value into lowercase.

replace_all(lowercase($.field), '\\s', '-')

Scopes

In the previous section, we have shown how to use the expression language to select a specific field. The selected field was part of our the current record being processed.

Actually, ScEL allows you to get access to additional fields through the used of scopes. Basically, a scope defined the root object on which a selector expression must evaluated.

The syntax to define an expression with a scope is of the form : “$<scope>.<selector expression string>”.

By default, if no scope is defined in the expression, the scope $value is implicitly used.

ScEL supports a number of predefined scopes that can be used for example :

  • To define the topic for the record.
  • To define the key for the record.
  • To get access to metadata about the source file.
  • Etc.
ScopeDescriptionType
$headersThe record headers-
$keyThe record keystring
$metadataThe file metadatastruct
$offsetThe offset information of this record into the source filestruct
$systemThe system environment variables and runtime propertiesstruct
$timestampThe record timestamplong
$topicThe output topicstring
$valueThe record valuestruct
$variablesThe contextual filter-chain variablesmap[string, object]

Note, that in case of failures more fields are added to the current filter context (see : Handling Failures)

Record Headers

The scope headers allows defining the headers of the output record.

Record key

The scope key allows defining the key of the output record. Only string key is currently supported.

Source Metadata

The scope metadata allows read access to information about the file being processing.

Predefined Fields (ScEL)DescriptionType
$metadata.nameThe file namestring
$metadata.pathThe file directory pathstring
$metadata.absolutePathThe file absolute pathstring
$metadata.hashThe file CRC32 hashint
$metadata.lastModifiedThe file last modified time.long
$metadata.sizeThe file sizelong
$metadata.inodeThe file Unix inodelong

Record Offset

The scope offset allows read access to information about the original position of the record into the source file. The available fields depend on the configured FileInputRecord.

Predefined Fields (ScEL)DescriptionType
$offset.timestampThe creation time of the record (millisecond)long

Information only available if RowFilterReader is configured.

Predefined Fields (ScEL)DescriptionType
$offset.startPositionThe start position of the record into the source filelong
$offset.endPositionThe end position of the record into the source filelong
$offset.sizeThe size in byteslong
$offset.rowThe row number of the record into the sourcelong

Information only available if BytesArrayInputReader is configured.

Predefined Fields (ScEL)DescriptionType
$offset.startPositionThe start position of the record into the source file (always equals to 0)long
$offset.endPositionThe end position of the record into the source file (equals to the file size)long

Information only available if AvroFilterInputReader is configured.

Predefined Fields (ScEL)DescriptionType
$offset.blockStartThe start position of the current blocklong
$offset.positionThe position into the current block.long
$offset.recordsThe number of record read into the current block.long

System

The scope system allows accessing to the system environment variables and runtime properties.

Predefined Fields (ScEL)DescriptionType
$system.envThe system environment variables.map[string, string]
$system.propsThe system environment properties.map[string, string]

Timestamp

The scope $timestamp allows defining the timestamp of the output record.

Topic

The scope $topic allows defining the target topic of the output record.

Value

The scope $value allows defining the fields of the output record

Variables

The scope $variables allows read/write access to a simple key-value map structure. This scope can be used to share user-defined variables between Processing Filters.

Note : variables are not cached between records.