Beam YAML Transform Index

AssertEqual

Asserts that the input contains exactly the elements provided.

This is primarily used for testing; it will cause the entire pipeline to fail if the input to this transform is not exactly the set of elements given in the config parameter.

As with Create, YAML/JSON-style mappings are interpreted as Beam rows, e.g.

type: AssertEqual
input: SomeTransform
config:
  elements:
     - {a: 0, b: "foo"}
     - {a: 1, b: "bar"}

would ensure that SomeTransform produced exactly two elements with values (a=0, b="foo") and (a=1, b="bar") respectively.

Configuration

Usage

type: AssertEqual
input: ...
config:
  elements:
  - element
  - element
  - ...

AssignTimestamps

Assigns a new timestamp each element of its input.

This can be useful when reading records that have the timestamp embedded in them, for example with various file types or other sources that by default set all timestamps to the infinite past.

Note that the timestamp should only be set forward, as setting it backwards may not cause it to hold back an already advanced watermark and the data could become droppably late.

Supported languages: generic, javascript, python

Configuration

Usage

type: AssignTimestamps
input: ...
config:
  timestamp: timestamp
  language: "language"
  error_handling:
    output: "output"

Combine

Groups and combines records sharing common fields.

Built-in combine functions are sum, max, min, all, any, mean, count, group, concat but custom aggregation functions can be used as well.

See also the documentation on YAML Aggregation.

Supported languages: calcite, generic, javascript, python, sql

Configuration

Usage

type: Combine
input: ...
config:
  group_by:
  - "group_by"
  - "group_by"
  - ...
  combine:
    a:
      a: combine_value_a_value_a
      b: combine_value_a_value_b
      c: ...
    b:
      a: combine_value_b_value_a
      b: combine_value_b_value_b
      c: ...
    c: ...
  language: "language"

Create

Creates a collection containing a specified set of elements.

This transform always produces schema'd data. For example

type: Create
config:
  elements: [1, 2, 3]

will result in an output with three elements with a schema of Row(element=int) whereas YAML/JSON-style mappings will be interpreted directly as Beam rows, e.g.

type: Create
config:
  elements:
     - {first: 0, second: {str: "foo", values: [1, 2, 3]}}
     - {first: 1, second: {str: "bar", values: [4, 5, 6]}}

will result in a schema of the form (int, Row(string, List[int])).

This can also be expressed as YAML

type: Create
config:
  elements:
    - first: 0
      second:
        str: "foo"
         values: [1, 2, 3]
    - first: 1
      second:
        str: "bar"
         values: [4, 5, 6]

Configuration

Usage

type: Create
config:
  elements:
  - element
  - element
  - ...
  reshuffle: true|false

Explode

Explodes (aka unnest/flatten) one or more fields producing multiple rows.

Given one or more fields of iterable type, produces multiple rows, one for each value of that field. For example, a row of the form ('a', [1, 2, 3]) would expand to ('a', 1), ('a', 2'), and ('a', 3) when exploded on the second field.

This is akin to a FlatMap when paired with the MapToFields transform.

See more complete documentation on YAML Mapping Functions.

Configuration

Usage

type: Explode
input: ...
config:
  fields: fields
  cross_product: true|false
  error_handling:
    a: error_handling_value_a
    b: error_handling_value_b
    c: ...

Filter

Keeps only records that satisfy the given criteria.

See more complete documentation on YAML Filtering.

Supported languages: calcite, generic, java, javascript, python, sql

Configuration

Usage

type: Filter
input: ...
config:
  keep: keep
  language: "language"
  error_handling:
    output: "output"

Flatten

Flattens multiple PCollections into a single PCollection.

The elements of the resulting PCollection will be the (disjoint) union of all the elements of all the inputs.

Note that in YAML transforms can always take a list of inputs which will be implicitly flattened.

Configuration

No configuration parameters.

Usage

type: Flatten
input: ...
config: ...

Join

Joins two or more inputs using a specified condition.

For example

type: Join
input:
  input1: SomeTransform
  input2: AnotherTransform
  input3: YetAnotherTransform
config:
  type: inner
  equalities:
    - input1: colA
      input2: colB
    - input2: colX
      input3: colY
  fields:
    input1: [colA, colB, colC]
    input2: {new_name: colB}

would perform an inner join on the three inputs satisfying the constraints that input1.colA = input2.colB and input2.colX = input3.colY emitting rows with colA, colB and colC from input1, the values of input2.colB as a field called new_name, and all the fields from input3.

Configuration

Usage

type: Join
input: ...
config:
  equalities: equalities
  type: type
  fields:
    a: fields_value_a
    b: fields_value_b
    c: ...

LogForTesting

Logs each element of its input PCollection.

The output of this transform is a copy of its input for ease of use in chain-style pipelines.

Configuration

Usage

type: LogForTesting
input: ...
config:
  level: "level"
  prefix: "prefix"

MLTransform

Configuration

Usage

type: MLTransform
input: ...
config:
  write_artifact_location: "write_artifact_location"
  read_artifact_location: "read_artifact_location"
  transforms:
  - transforms
  - transforms
  - ...

MapToFields

Creates records with new fields defined in terms of the input fields.

See more complete documentation on YAML Mapping Functions.

Supported languages: calcite, generic, java, javascript, python, sql

Configuration

Usage

type: MapToFields
input: ...
config:
  language: language
  error_handling:
    output: "output"
  mapping_args: mapping_args

Partition

Splits an input into several distinct outputs.

Each input element will go to a distinct output based on the field or function given in the by configuration parameter.

Supported languages: generic, javascript, python

Configuration

Usage

type: Partition
input: ...
config:
  by: by
  outputs:
  - "outputs"
  - "outputs"
  - ...
  unknown_output: "unknown_output"
  error_handling:
    a: error_handling_value_a
    b: error_handling_value_b
    c: ...
  language: "language"

PyTransform

A Python PTransform identified by fully qualified name.

This allows one to import, construct, and apply any Beam Python transform. This can be useful for using transforms that have not yet been exposed via a YAML interface. Note, however, that conversion may be required if this transform does not accept or produce Beam Rows.

For example

type: PyTransform
config:
   constructor: apache_beam.pkg.mod.SomeClass
   args: [1, 'foo']
   kwargs:
     baz: 3

can be used to access the transform apache_beam.pkg.mod.SomeClass(1, 'foo', baz=3).

See also the documentation on Inlining Python.

Configuration

Usage

type: PyTransform
input: ...
config:
  constructor: "constructor"
  args:
  - arg
  - arg
  - ...
  kwargs:
    a: kwargs_value_a
    b: kwargs_value_b
    c: ...

Sql

Configuration

Usage

type: Sql
input: ...
config: ...

WindowInto

A window transform assigning windows to each element of a PCollection.

The assigned windows will affect all downstream aggregating operations, which will aggregate by window as well as by key.

See the Beam documentation on windowing for more details.

Sizes, offsets, periods and gaps (where applicable) must be defined using a time unit suffix 'ms', 's', 'm', 'h' or 'd' for milliseconds, seconds, minutes, hours or days, respectively. If a time unit is not specified, it will default to 's'.

For example

windowing:
   type: fixed
   size: 30s

Note that any Yaml transform can have a windowing parameter, which is applied to its inputs (if any) or outputs (if there are no inputs) which means that explicit WindowInto operations are not typically needed.

Configuration

Usage

type: WindowInto
input: ...
config:
  windowing: windowing

ReadFromAvro

A PTransform for reading records from avro files.

Each record of the resulting PCollection will contain a single record read from a source. Records that are of simple types will be mapped to beam Rows with a single record field containing the records value. Records that are of Avro type RECORD will be mapped to Beam rows that comply with the schema contained in the Avro file that contains those records.

Configuration

Usage

type: ReadFromAvro
config:
  path: path

WriteToAvro

A PTransform for writing avro files.

If the input has a schema, a corresponding avro schema will be automatically generated and used to write the output records.

Configuration

Usage

type: WriteToAvro
input: ...
config:
  path: path

ReadFromBigQuery

Reads data from BigQuery.

Exactly one of table or query must be set. If query is set, neither row_restriction nor fields should be set.

Configuration

Usage

type: ReadFromBigQuery
config:
  table: "table"
  query: "query"
  row_restriction: "row_restriction"
  fields:
  - "field"
  - "field"
  - ...

WriteToBigQuery

Configuration

Usage

type: WriteToBigQuery
input: ...
config: ...

ReadFromCsv

A PTransform for reading comma-separated values (csv) files into a PCollection.

Configuration

Usage

type: ReadFromCsv
config:
  path: "path"
  delimiter: delimiter
  comment: comment

WriteToCsv

A PTransform for writing a schema'd PCollection as a (set of) comma-separated values (csv) files.

Configuration

Usage

type: WriteToCsv
input: ...
config:
  path: "path"
  delimiter: delimiter

ReadFromJdbc

Configuration

Usage

type: ReadFromJdbc
config: ...

WriteToJdbc

Configuration

Usage

type: WriteToJdbc
input: ...
config: ...

ReadFromJson

A PTransform for reading json values from files into a PCollection.

Configuration

Usage

type: ReadFromJson
config:
  path: "path"

WriteToJson

A PTransform for writing a PCollection as json values to files.

Configuration

Usage

type: WriteToJson
input: ...
config:
  path: "path"

ReadFromKafka

Configuration

Usage

type: ReadFromKafka
config: ...

WriteToKafka

Configuration

Usage

type: WriteToKafka
input: ...
config: ...

ReadFromMySql

Configuration

Usage

type: ReadFromMySql
config: ...

WriteToMySql

Configuration

Usage

type: WriteToMySql
input: ...
config: ...

ReadFromOracle

Configuration

Usage

type: ReadFromOracle
config: ...

WriteToOracle

Configuration

Usage

type: WriteToOracle
input: ...
config: ...

ReadFromParquet

A PTransform for reading Parquet files.

Configuration

Usage

type: ReadFromParquet
config:
  path: path

WriteToParquet

A PTransform for writing parquet files.

Configuration

Usage

type: WriteToParquet
input: ...
config:
  path: path

ReadFromPostgres

Configuration

Usage

type: ReadFromPostgres
config: ...

WriteToPostgres

Configuration

Usage

type: WriteToPostgres
input: ...
config: ...

ReadFromPubSub

Reads messages from Cloud Pub/Sub.

Configuration

Usage

type: ReadFromPubSub
config:
  topic: "topic"
  subscription: "subscription"
  format: "format"
  schema: schema
  attributes:
  - "attribute"
  - "attribute"
  - ...
  attributes_map: "attributes_map"
  id_attribute: "id_attribute"
  timestamp_attribute: "timestamp_attribute"
  error_handling:
    output: "output"

WriteToPubSub

Writes messages to Cloud Pub/Sub.

Configuration

Usage

type: WriteToPubSub
input: ...
config:
  topic: "topic"
  format: "format"
  schema: schema
  attributes:
  - "attribute"
  - "attribute"
  - ...
  attributes_map: "attributes_map"
  id_attribute: "id_attribute"
  timestamp_attribute: "timestamp_attribute"
  error_handling:
    output: "output"

ReadFromPubSubLite

Configuration

Usage

type: ReadFromPubSubLite
config: ...

WriteToPubSubLite

Configuration

Usage

type: WriteToPubSubLite
input: ...
config: ...

ReadFromSpanner

Configuration

Usage

type: ReadFromSpanner
config: ...

WriteToSpanner

Configuration

Usage

type: WriteToSpanner
input: ...
config: ...

ReadFromSqlServer

Configuration

Usage

type: ReadFromSqlServer
config: ...

WriteToSqlServer

Configuration

Usage

type: WriteToSqlServer
input: ...
config: ...

ReadFromText

Reads lines from a text files.

The resulting PCollection consists of rows with a single string filed named "line."

Configuration

Usage

type: ReadFromText
config:
  path: "path"

WriteToText

Writes a PCollection to a (set of) text files(s).

The input must be a PCollection whose schema has exactly one field.

Configuration

Usage

type: WriteToText
input: ...
config:
  path: "path"