Performs anomaly detection on a PCollection of data.
This PTransform applies an AnomalyDetector or EnsembleAnomalyDetector to
the input data and returns a PCollection of AnomalyResult objects.
Examples
# Run a single anomaly detector
p | AnomalyDetection(ZScore(features=["x1"]))
# Run an ensemble anomaly detector
sub_detectors = [ZScore(features=["x1"]), IQR(features=["x2"])]
p | AnomalyDetection(EnsembleAnomalyDetector(
sub_detectors, aggregation_strategy=AnyVote()))
? (Optional) : The AnomalyDetector or EnsembleAnomalyDetector to use.type: AnomalyDetection
input: ...
config:
detector: detector
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.
Array[?] : The set of elements that should belong to the PCollection.
YAML/JSON-style mappings will be interpreted as Beam rows.type: AssertEqual
input: ...
config:
elements:
- element
- element
- ...
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.
timestamp ? (Optional) : A field, callable, or expression giving the new timestamp.
language string (Optional) : The language of the timestamp expression.
error_handling Row (Optional) : Whether and how to handle errors during timestamp
evaluation.
Row fields:
string : Name to use for the output error collectiontype: AssignTimestamps
input: ...
config:
timestamp: timestamp
language: "language"
error_handling:
output: "output"
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.
group_by Array[string] : The field(s) to aggregate on.
combine Map[string, Map[string, ?]] : The aggregation function to use.
language string (Optional) : The language used to define (and execute) the
custom callables in combine. Defaults to generic.
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"
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]
If the elements are a mix of dicts and non-dicts, the non-dict elements will be wrapped in a Row with a single field "element". For example
type: Create
config:
elements: [1, {"a": 2}]
will result in an output with two elements with a schema of Row(element=int, a=int) looking like:
Row(element=1, a=None)
Row(element=None, a=2)
elements Array[?] : The set of elements that should belong to the PCollection.
YAML/JSON-style mappings will be interpreted as Beam rows.
Primitives will be mapped to rows with a single "element" field.
reshuffle boolean (Optional) : (optional) Whether to introduce a reshuffle (to possibly
redistribute the work) if there is more than one element in the
collection. Defaults to True.
type: Create
config:
elements:
- element
- element
- ...
reshuffle: true|false
Searches a Google Cloud Healthcare DICOM store using QIDO-RS.
This transform takes an input PCollection of Rows describing QIDO search requests and returns Rows with the search results encoded as JSON.
Each input Row must include:
studies, series, or instances.Successful outputs are Rows with:
Failed searches raise unless error_handling is set, in which case they
are routed to the configured error output.
buffer_size int64 : Number of requests to buffer before flushing.
max_workers int64 : Maximum number of threads used to issue requests.
error_handling Row (Optional) : If specified, should be a mapping giving an output into
which to emit failed searches, as described at
https://beam.apache.org/documentation/sdks/yaml-errors/
Row fields:
string : Name to use for the output error collectiontype: DicomSearch
input: ...
config:
buffer_size: buffer_size
max_workers: max_workers
error_handling: error_handling
The Enrichment transform allows one to dynamically enhance elements in a pipeline by performing key-value lookups against external services like APIs or databases.
Example using BigTable:
- type: Enrichment
config:
enrichment_handler: 'BigTable'
handler_config:
project_id: 'apache-beam-testing'
instance_id: 'beam-test'
table_id: 'bigtable-enrichment-test'
row_key: 'product_id'
timeout: 30
For more information on Enrichment, see the Beam docs.
enrichment_handler string : Specifies the source from where data needs
to be extracted into the pipeline for enriching data. One of
"BigQuery", "BigTable", "FeastFeatureStore" or "VertexAIFeatureStore".
handler_config Map[string, ?] : Specifies the parameters for the respective
enrichment_handler in a YAML/JSON format. To see the full set of
handler_config parameters, see their corresponding doc pages:
timeout double (Optional) : Timeout for source requests in seconds. Defaults to 30
seconds.
type: Enrichment
input: ...
config:
enrichment_handler: "enrichment_handler"
handler_config:
a: handler_config_value_a
b: handler_config_value_b
c: ...
timeout: timeout
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.
fields ? (Optional) : The list of fields to expand.
cross_product boolean (Optional) : If multiple fields are specified, indicates whether the
full cross-product of combinations should be produced, or if the
first element of the first field corresponds to the first element
of the second field, etc. For example, the row
(['a', 'b'], [1, 2]) would expand to the four rows
('a', 1), ('a', 2), ('b', 1), and ('b', 2) when
cross_product is set to true but only the two rows
('a', 1) and ('b', 2) when it is set to false.
Only meaningful (and required) if multiple rows are specified.
error_handling Row (Optional) : Whether and how to handle errors during iteration.
Row fields:
string : Name to use for the output error collectiontype: Explode
input: ...
config:
fields: fields
cross_product: true|false
error_handling:
a: error_handling_value_a
b: error_handling_value_b
c: ...
Extracts the implicit windowing information from an element and makes it explicit as field(s) in the element itself.
The following windowing parameter values are supported:
timestamp: The event timestamp of the current element.window_start: The start of the window iff it is an interval window.window_end: The (exclusive) end of the window.window_string: The string representation of the window.window_type: The type of the window as a string.winodw_object: The actual window object itself,
as a Java or Python object.pane_info: A schema'd representation of the current pane info, including
its index, whether it was the last firing, etc.As a convenience, a list rather than a mapping of fields may be provided, in which case the fields will be named according to the requested values.
Map[string, string] (Optional) : A mapping of new field names to various windowing parameters,
as documented above. If omitted, defaults to
[timestamp, window_start, window_end].type: ExtractWindowingInfo
input: ...
config:
fields:
a: "fields_value_a"
b: "fields_value_b"
c: ...
Keeps only records that satisfy the given criteria.
See more complete documentation on YAML Filtering.
Supported languages: calcite, generic, java, javascript, python, sql.
language string (Optional)
keep Row
Row fields:
callable string (Optional) : Source code of a public class implementing Function
expression string (Optional) : Source code of a java expression in terms of the schema fields.
name string (Optional) : Fully qualified name of either a class implementing Function
path string (Optional) : Path to a jar file implementing the function referenced in name.
error_handling Row (Optional) : This option specifies whether and where to output error rows.
Row fields:
string : Name to use for the output error collectiontype: Filter
input: ...
config:
language: "language"
keep:
callable: "callable"
expression: "expression"
name: "name"
path: "path"
error_handling:
output: "output"
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.
No configuration parameters.
type: Flatten
input: ...
table string : A fully-qualified table identifier.
catalog_properties Map[string, string] (Optional) : Properties used to set up the Iceberg catalog.
config_properties Map[string, string] (Optional) : Properties passed to the Hadoop configuration the catalog uses.
triggering_frequency_seconds int32 (Optional) : For a streaming pipeline, sets the frequency at which incoming files are appended (default 600, or 10min).
manifest_file_size int32 (Optional) : The number of data files per manifest (default 10,000 files).
location_prefix string (Optional) : The prefix shared among all partitions. For example, a data file may have the following location:%n'gs://bucket/namespace/table/data/id=13/name=beam/data_file.parquet'%n%nThe provided prefix should go up until the partition information:%n'gs://bucket/namespace/table/data/'.%nIf not provided, will try determining each DataFile's partition from its metrics metadata.
partition_fields Array[string] (Optional) : Fields used to create a partition spec that is applied when tables are created. For a field 'foo', the available partition transforms are:
footruncate(foo, N)bucket(foo, N)hour(foo)day(foo)month(foo)year(foo)void(foo)For more information on partition transforms, please visit https://iceberg.apache.org/spec/#partition-transforms.
table_properties Map[string, string] (Optional) : Iceberg table properties to be set on the table when it is created.
For more information on table properties, please visit https://iceberg.apache.org/docs/latest/configuration/#table-properties.
sort_fields Array[string] (Optional) : Fields used to set the table's sort order, applied when the table is created. Each entry has the form <term> [asc|desc] [nulls first|nulls last], where <term> is a field name or one of the partition transforms (e.g. bucket(col, 4), day(ts)). Direction defaults to ascending; null order defaults to nulls-first for ascending and nulls-last for descending.
For more information on sort orders, please visit https://iceberg.apache.org/spec/#sort-orders.
schema_evolution_options Array[string] (Optional) : Lets the transform change the table schema so that the table has a column for every column the files have. Values: ALLOW_FIELD_ADDITION (columns a file has and the table lacks are added, as optional), ALLOW_FIELD_RELAXATION (a required table column becomes optional when a file lacks it or may hold nulls in it), ALLOW_TYPE_PROMOTION (a column type is widened, for example int to long). Leave it empty to never change the table schema. When any option is set, the transform reads the footer of every Parquet file and commits the allowed changes before registering any file, so every registered file has statistics for all of its columns. A file that needs a change that is not allowed is incompatible; see incompatible_schema_handling. Only Parquet files can be checked: ORC and Avro files are sent to the error output unless unverifiable_file_handling is ACCEPT. Files sent to the error output are dropped unless error_handling is set. If the table does not exist it is created from the union of the Parquet schemas; with none, every file goes to the error output. Batch pipelines only; streaming pipelines cannot use schema evolution yet.
required_columns Array[string] (Optional) : Columns that must always be present and never null, as dotted paths for nested fields (for example address.city). They are never made optional, whatever the options allow, and are created as required when the transform creates the table. A file that lacks one of these columns, or holds nulls in it, is sent to the error output (see error_handling). So is a file whose footer marks the column as optional and has no null-count statistics for it, unless unverifiable_file_handling is ACCEPT. Requires schema_evolution_options.
incompatible_schema_handling string (Optional) : What happens when a file's schema cannot be made to fit the table: it needs a change that is not allowed, or it conflicts with the table or with another file. FAIL_PIPELINE (the default) fails the pipeline before any schema change is committed. ROUTE_TO_ERRORS commits the changes for the other files and sends the incompatible files to the error output; it requires error_handling.
unverifiable_file_handling string (Optional) : What happens to a file the checks cannot verify: an ORC or Avro file (the checks read Parquet footers only), or a Parquet file with no null-count statistics for a required column (statistics disabled by the writer, or a column under a list or map). REJECT (the default) sends the file to the error output (see error_handling). ACCEPT registers it without the checks; such files are counted and logged. A file that fails a check is always sent to the error output. An accepted file that lacks a required column, or holds nulls in it, makes reads of the table fail.
dry_run boolean (Optional) : When true, nothing is committed or registered: the transform reads the files' schemas and emits a dry_run_report output with one row that describes what a real run would do. Its allowed field is true when every file schema can be merged and the configuration raises no problem; otherwise its reason field says what a real run would do about it (fail, or route the files to the error output). Its schemas field lists each distinct file schema with the changes a real run would make for it and, when it cannot be merged, why. The output only exists when this is set; consume it as input: <this transform's name>.dry_run_report. Against a missing table, a REST catalog needs table-create permission even though no table is created.
error_handling Row (Optional) : Whether and where to output the files that could not be registered, as rows with the file path and the error. Without it those files are dropped.
Row fields:
string : Name to use for the output error collectiontype: IcebergAddFiles
config:
table: "table"
catalog_properties:
a: "catalog_properties_value_a"
b: "catalog_properties_value_b"
c: ...
config_properties:
a: "config_properties_value_a"
b: "config_properties_value_b"
c: ...
triggering_frequency_seconds: triggering_frequency_seconds
manifest_file_size: manifest_file_size
location_prefix: "location_prefix"
partition_fields:
- "partition_fields"
- "partition_fields"
- ...
table_properties:
a: "table_properties_value_a"
b: "table_properties_value_b"
c: ...
sort_fields:
- "sort_fields"
- "sort_fields"
- ...
schema_evolution_options:
- "schema_evolution_options"
- "schema_evolution_options"
- ...
required_columns:
- "required_columns"
- "required_columns"
- ...
incompatible_schema_handling: "incompatible_schema_handling"
unverifiable_file_handling: "unverifiable_file_handling"
dry_run: true|false
error_handling:
output: "output"
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.
equalities ? (Optional) : The condition to join on. A list of sets of columns that should
be equal to fulfill the join condition. For the simple scenario of
joining on the same column across all inputs where the column name is
the same, one can specify the column name as the equality rather than
having to list it for every input.
type ? (Optional) : The type of join. Could be a string value in
["inner", "left", "right", "outer"] that specifies the type of join to
be performed. For scenarios with multiple inputs to join where different
join types are desired, specify the inputs to be outer joined. For
example, {outer: [input1, input2]} means that input1 and input2
will be outer joined using the conditions specified, while other inputs
will be inner joined.
fields Map[string, ?] (Optional) : The fields to be outputted. A mapping with the input alias as the
key and the list of fields in the input to be outputted.
The value in the map
can either be a dictionary with the new field name as the key and the
original field name as the value (e.g new_field_name: field_name), or a
list of the fields to be outputted with their original names
(e.g [col1, col2, col3]), or an '*' indicating all fields in the
input will be outputted. If not specified, all fields from all inputs
will be outputted.
type: Join
input: ...
config:
equalities: equalities
type: type
fields:
a: fields_value_a
b: fields_value_b
c: ...
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.
level string (Optional) : one of ERROR, INFO, or DEBUG, mapped to a corresponding
language-specific logging level
prefix string (Optional) : an optional identifier that will get prepended to the element
being logged
error_handling Row (Optional) : This option specifies whether and where to output error rows.
Row fields:
string : Name to use for the output error collectiontype: LogForTesting
input: ...
config:
level: "level"
prefix: "prefix"
error_handling:
output: "output"
write_artifact_location string (Optional)
read_artifact_location string (Optional)
transforms Array[?] (Optional)
type: MLTransform
input: ...
config:
write_artifact_location: "write_artifact_location"
read_artifact_location: "read_artifact_location"
transforms:
- transforms
- transforms
- ...
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.
fields Map[string, ?] : The output fields to compute, each mapping to the expression or
callable that creates them.
append boolean (Optional) : Whether to append the created fields to the set of
fields already present, outputting a union of both the new fields and
the original fields for each record. Defaults to False.
drop Array[string] (Optional) : If append is true, enumerates a subset of fields from the
original record that should not be kept
language string (Optional) : The language used to define (and execute) the
expressions and/or callables in fields. Defaults to generic.
dependencies Array[string] (Optional) : An optional list of extra dependencies that are needed for
these UDFs. The interpretation of these strings is language-dependent.
error_handling Row (Optional) : Whether and where to output records that throw errors when
the above expressions are evaluated.
Row fields:
string : Name to use for the output error collectiontype: MapToFields
input: ...
config:
language: "language"
append: true|false
drop:
- "drop"
- "drop"
- ...
fields:
a:
expression: "expression"
callable: "callable"
name: "name"
path: "path"
b:
expression: "expression"
callable: "callable"
name: "name"
path: "path"
c: ...
error_handling:
output: "output"
Matches file patterns from the input PCollection.
This transform returns a PCollection of matching files, each represented as a Row with path, size_in_bytes, and last_updated_in_seconds fields.
file_pattern string (Optional) : The name of the field in the input PCollection that contains
the file pattern string. If not specified and the input PCollection has
exactly one field, that field will be used.
empty_match_treatment string : How to treat empty matches. Possible values are
'ALLOW', 'DISALLOW', and 'ALLOW_IF_WILDCARD'. Defaults to 'ALLOW'.
type: MatchAll
input: ...
config:
file_pattern: "file_pattern"
empty_match_treatment: "empty_match_treatment"
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.
by ? (Optional) : A field, callable, or expression giving the destination output for
this element. Should return a string that is a member of the outputs
parameter. If unknown_output is also set, other returns values are
accepted as well, otherwise an error will be raised.
outputs Array[string] : The set of outputs into which this input is being partitioned.
unknown_output string (Optional) : (Optional) If set, indicates a destination output for any
elements that are not assigned an output listed in the outputs
parameter.
error_handling Row (Optional) : (Optional) Whether and how to handle errors during
partitioning.
Row fields:
string : Name to use for the output error collectionlanguage string : The language of the by expression.
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"
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.
constructor string : Fully qualified name of a callable used to construct the
transform. Often this is a class such as
apache_beam.pkg.mod.SomeClass but it can also be a function or
any other callable that returns a PTransform.
args Array[?] (Optional) : A list of parameters to pass to the callable as positional
arguments.
kwargs Map[string, ?] (Optional) : A list of parameters to pass to the callable as keyword
arguments.
type: PyTransform
input: ...
config:
constructor: "constructor"
args:
- arg
- arg
- ...
kwargs:
a: kwargs_value_a
b: kwargs_value_b
c: ...
Reshuffles the elements of a PCollection.
Redistributes the elements of a PCollection to prevent fusion or balance load.
int64 (Optional) : (optional) Specifies the maximum random keys that would be
generated. If not set, a default value is used.type: Reshuffle
input: ...
config:
num_buckets: num_buckets
A transform that takes the input rows, containing examples (or features), for use on an ML model. The transform then appends the inferences (or predictions) for those examples to the input row.
A ModelHandler must be passed to the model_handler parameter. The
ModelHandler is responsible for configuring how the ML model will be loaded
and how input data will be passed to it. Every ModelHandler has a config tag,
similar to how a transform is defined, where the parameters are defined.
For example:
- type: RunInference
config:
model_handler:
type: ModelHandler
config:
param_1: arg1
param_2: arg2
...
By default, the RunInference transform will return the
input row with a single field appended named by the inference_tag parameter
("inference" by default) that contains the inference directly returned by the
underlying ModelHandler, after any optional postprocessing.
For example, if the input had the following:
Row(question="What is a car?")
The output row would look like:
Row(question="What is a car?", inference=...)
where the inference tag can be overridden with the inference_tag
parameter.
However, if one specified the following transform config:
- type: RunInference
config:
inference_tag: my_inference
model_handler: ...
The output row would look like:
Row(question="What is a car?", my_inference=...)
See more complete documentation on the underlying RunInference transform.
In most cases, the model will be expecting data in a particular data format,
whether it be a Python Dict, PyTorch tensor, etc. However, the outputs of all
built-in Beam YAML transforms are Beam Rows. To allow for transforming
the Beam Row into a data format the model recognizes, each ModelHandler is
equipped with a preprocessing parameter for performing necessary data
preprocessing. It is possible for a ModelHandler to define a default
preprocessing function, but in most cases, one will need to be specified by
the caller.
For example, using callable:
pipeline:
type: chain
transforms:
- type: Create
config:
elements:
- question: "What is a car?"
- question: "Where is the Eiffel Tower located?"
- type: RunInference
config:
model_handler:
type: ModelHandler
config:
param_1: arg1
param_2: arg2
preprocess:
callable: 'lambda row: {"prompt": row.question}'
...
In the above example, the Create transform generates a collection of two Beam Row elements, each with a single field - "question". The model, however, expects a Python Dict with a single key, "prompt". In this case, we can specify a simple Lambda function (alternatively could define a full function), to map the data.
It is also possible to define a postprocessing function to postprocess the
data output by the ModelHandler. See the documentation for the ModelHandler
you intend to use (list defined below under model_handler parameter doc).
In many cases, before postprocessing, the object will be a PredictionResult. # pylint: disable=line-too-long This type behaves very similarly to a Beam Row and fields can be accessed using dot notation. However, make sure to check the docs for your ModelHandler to see which fields its PredictionResult contains or if it returns a different object altogether.
For example:
- type: RunInference
config:
model_handler:
type: ModelHandler
config:
param_1: arg1
param_2: arg2
postprocess:
callable: |
def fn(x: PredictionResult):
return beam.Row(x.example, x.inference, x.model_id)
...
The above example demonstrates converting the original output data type (in this case it is PredictionResult), and converts to a Beam Row, which allows for easier mapping in a later transform.
For both preprocessing and postprocessing, it is also possible to specify a
Python UDF (User-defined function) file that contains the function. This is
possible by specifying the path to the file (local file or GCS path) and
the name of the function in the file.
For example:
- type: RunInference
config:
model_handler:
type: ModelHandler
config:
param_1: arg1
param_2: arg2
preprocess:
path: gs://my-bucket/path/to/preprocess.py
name: my_preprocess_fn
postprocess:
path: gs://my-bucket/path/to/postprocess.py
name: my_postprocess_fn
...
model_handler Map[string, ?] : Specifies the parameters for the respective
model_handler in a YAML/JSON format. To see the full set of
handler_config parameters, see their corresponding doc pages:
inference_tag string (Optional) : The tag to use for the returned inference. Default is
'inference'.
inference_args Map[string, ?] (Optional) : Extra arguments for models whose inference call requires
extra parameters. Make sure to check the underlying ModelHandler docs to
see which args are allowed.
type: RunInference
input: ...
config:
model_handler:
a: model_handler_value_a
b: model_handler_value_b
c: ...
inference_tag: "inference_tag"
inference_args:
a: inference_args_value_a
b: inference_args_value_b
c: ...
type: Sql
input: ...
config: ...
Strips error metadata from outputs returned via error handling.
Generally the error outputs for transformations return information about the error encountered (e.g. error messages and tracebacks) in addition to the failing element itself. This transformation attempts to remove that metadata and returns the bad element alone which can be useful for re-processing.
For example, in the following pipeline snippet
- name: MyMappingTransform
type: MapToFields
input: SomeInput
config:
language: python
fields:
...
error_handling:
output: errors
- name: RecoverOriginalElements
type: StripErrorMetadata
input: MyMappingTransform.errors
the output of RecoverOriginalElements will contain exactly those elements
from SomeInput that failed to processes (whereas MyMappingTransform.errors
would contain those elements paired with error information).
Note that this relies on the preceding transform actually returning the failing input in a schema'd way. Most built-in transformation follow the correct conventions.
No configuration parameters.
type: StripErrorMetadata
input: ...
Validates each element of a PCollection against a json schema.
schema Map[string, ?] : A json schema against which to validate each element.
error_handling Row (Optional) : Whether and how to handle errors during iteration.
If this is not set, invalid elements will fail the pipeline, otherwise
invalid elements will be passed to the specified error output along
with information about how the schema was invalidated.
Row fields:
string : Name to use for the output error collectiontype: ValidateWithSchema
input: ...
config:
schema:
a: schema_value_a
b: schema_value_b
c: ...
error_handling:
a: error_handling_value_a
b: error_handling_value_b
c: ...
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.
? (Optional) : the type and parameters of the windowing to performtype: WindowInto
input: ...
config:
windowing: windowing
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.
? (Optional)type: ReadFromAvro
config:
path: path
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.
? (Optional)type: WriteToAvro
input: ...
config:
path: path
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.
table string (Optional) : The table to read from, specified as DATASET.TABLE
or PROJECT:DATASET.TABLE.
query string (Optional) : A query to be used instead of the table argument.
row_restriction string (Optional) : Optional SQL text filtering statement, similar to a
WHERE clause in a query. Aggregates are not supported. Restricted to a
maximum length for 1 MB.
fields Array[string] (Optional)
schema ? (Optional) : Required when query is set. A BigQuery schema describing
the query result columns, e.g.
{'fields': [{'name': 'col', 'type': 'STRING', 'mode': 'NULLABLE'}]}.
Not applicable when reading from a table (schema is auto-derived).
type: ReadFromBigQuery
config:
table: "table"
query: "query"
row_restriction: "row_restriction"
fields:
- "field"
- "field"
- ...
schema: schema
table string : The bigquery table to write to. Format: [${PROJECT}:]${DATASET}.${TABLE}
create_disposition string (Optional) : Optional field that specifies whether the job is allowed to create new tables. The following values are supported: CREATE_IF_NEEDED (the job may create the table), CREATE_NEVER (the job must fail if the table does not exist already).
write_disposition string (Optional) : Specifies the action that occurs if the destination table already exists. The following values are supported: WRITE_TRUNCATE (overwrites the table data), WRITE_APPEND (append the data to the table), WRITE_EMPTY (job must fail if the table is not empty).
error_handling Row (Optional) : This option specifies whether and where to output unwritable rows.
Row fields:
string : Name to use for the output error collectionnum_streams int32 (Optional) : Specifies the number of write streams that the Storage API sink will use.
type: WriteToBigQuery
config:
table: "table"
create_disposition: "create_disposition"
write_disposition: "write_disposition"
error_handling:
output: "output"
num_streams: num_streams
Reads data from a Google Cloud Bigtable table. The transform requires the project ID, instance ID, and table ID parameters. Optionally, the output can be flattened or nested rows. Example usage: - type: ReadFromBigTable config: project: "my-gcp-project" instance: "my-bigtable-instance" table: "my-table"
project string : Google Cloud project ID containing the Bigtable instance.
instance string : Bigtable instance ID to connect to.
table string : Bigtable table ID to read from.
flatten boolean (Optional) : If set to false, output rows are nested; if true or omitted, output rows are flattened.
type: ReadFromBigTable
config:
project: "project"
instance: "instance"
table: "table"
flatten: true|false
Writes data to a Google Cloud Bigtable table. This transform requires the Google Cloud project ID, Bigtable instance ID, and table ID. The input PCollection should be schema-compliant mutations or keyed rows. Example usage: - type: WriteToBigTable input: input config: project: "my-gcp-project" instance: "my-bigtable-instance" table: "my-table"
project string : Google Cloud project ID containing the Bigtable instance.
instance string : Bigtable instance ID where the table is located.
table string : Bigtable table ID to write data into.
type: WriteToBigTable
input: ...
config:
project: "project"
instance: "instance"
table: "table"
A PTransform for reading comma-separated values (csv) files into a PCollection.
path string : The file path to read from. The path can contain glob
characters such as * and ?.
delimiter ? (Optional)
comment ? (Optional)
filename_column string (Optional) : If not None, the name of the column to add
to each record, containing the filename of the source file.
type: ReadFromCsv
config:
path: "path"
delimiter: delimiter
comment: comment
filename_column: "filename_column"
A PTransform for writing a schema'd PCollection as a (set of) comma-separated values (csv) files.
path string : The file path to write to. The files written will
begin with this prefix, followed by a shard identifier (see
num_shards) according to the file_naming parameter.
delimiter ? (Optional)
type: WriteToCsv
input: ...
config:
delimiter: "delimiter"
path: "path"
url string : The Datadog API URL.
api_key string : The Datadog API key.
min_batch_count int32 (Optional) : The minimum number of events to batch together for each write.
batch_count int32 (Optional) : The number of events to batch together for each write.
max_buffer_size int64 (Optional) : The maximum buffer size in bytes.
parallelism int32 (Optional) : The degree of parallelism for writing.
error_handling Row (Optional) : Specifies how to handle errors.
Row fields:
string : Name to use for the output error collectiontype: WriteToDatadog
input: ...
config:
url: "url"
api_key: "api_key"
min_batch_count: min_batch_count
batch_count: batch_count
max_buffer_size: max_buffer_size
parallelism: parallelism
error_handling:
output: "output"
connector string : Debezium connector type. Supported values: MYSQL, POSTGRES, SQLSERVER, ORACLE, and DB2.
username string : Username used to connect to the source database.
password string : Password used to connect to the source database.
host string : Hostname of the source database.
port int32 : Port of the source database.
table string : Fully qualified table name included in the Debezium change stream.
connection_properties Array[string] (Optional) : Additional Debezium connection properties in key=value format.
max_number_of_records int32 (Optional) : Maximum number of records to read before stopping.
max_time_to_run int64 (Optional) : Maximum time in milliseconds to run before stopping.
type: ReadFromDebezium
config:
connector: "connector"
username: "username"
password: "password"
host: "host"
port: port
table: "table"
connection_properties:
- "connection_properties"
- "connection_properties"
- ...
max_number_of_records: max_number_of_records
max_time_to_run: max_time_to_run
Reads a Delta Lake table.
table string : Identifier of the Delta Lake table.
version int64 (Optional) : Version of the Delta Lake table to read.
timestamp string (Optional) : Timestamp of the Delta Lake table to read.
hadoop_config Map[string, string] (Optional) : Hadoop configuration properties.
type: ReadFromDelta
config:
table: "table"
version: version
timestamp: "timestamp"
hadoop_config:
a: "hadoop_config_value_a"
b: "hadoop_config_value_b"
c: ...
Reads change records from a Delta Lake table.
table string : Identifier of the Delta Lake table.
start_version int64 (Optional) : Start version of the Delta Lake table to read changes from.
Either this or start_timestamp has to be provided.
start_timestamp string (Optional) : Start timestamp of the Delta Lake table to read changes
from. Should be specified in the ISO 8601 standard. Either this or
start_version has to be provided.
end_version int64 (Optional) : End version of the Delta Lake table to read changes up to.
end_timestamp string (Optional) : End timestamp of the Delta Lake table to read changes up to.
Should be specified in the ISO 8601 standard.
include_metadata_columns Array[string] (Optional) : Metadata columns to include in the output rows.
Supported columns are: _change_type, _commit_version, and _commit_timestamp.
hadoop_config Map[string, string] (Optional) : Properties passed to the Hadoop Configuration.
type: ReadFromDeltaCDC
config:
table: "table"
start_version: start_version
start_timestamp: "start_timestamp"
end_version: end_version
end_timestamp: "end_timestamp"
include_metadata_columns:
- "include_metadata_columns"
- "include_metadata_columns"
- ...
hadoop_config:
a: "hadoop_config_value_a"
b: "hadoop_config_value_b"
c: ...
Reads documents from a Google Cloud Firestore collection and outputs Beam Rows.
project string (Optional) : GCP project id. Defaults to GcpOptions project when unset.
database string (Optional) : Firestore database id. Defaults to FirestoreOptions firestoreDb when unset.
collection string : Firestore collection id to read from.
schema string : The schema in which the data is encoded, defined with JSON-schema syntax (https://json-schema.org/).
error_handling Row (Optional) : This option specifies whether and where to output rows that failed to be read.
Row fields:
string : Name to use for the output error collectiontype: ReadFromFirestore
config:
project: "project"
database: "database"
collection: "collection"
schema: "schema"
error_handling:
output: "output"
Writes Beam Rows to a Google Cloud Firestore collection.
project string (Optional) : GCP project id. Defaults to GcpOptions project when unset.
database string (Optional) : Firestore database id. Defaults to FirestoreOptions firestoreDb when unset.
collection string : Firestore collection id to write to.
document_id_field string (Optional) : Row field containing the document id. Defaults to document_id when unset.
error_handling Row (Optional) : This option specifies whether and where to output unwritable rows. Error handling is limited to data conversion failures before sending writes to Firestore.
Row fields:
string : Name to use for the output error collectiontype: WriteToFirestore
input: ...
config:
project: "project"
database: "database"
collection: "collection"
document_id_field: "document_id_field"
error_handling:
output: "output"
Reads messages from a JMS broker and outputs each message as a single payload (string) field.
By default the read is unbounded (streaming): it keeps consuming messages from the specified queue or topic until the pipeline is stopped. Setting maxNumRecords and/or maxReadTimeSeconds bounds the read, producing a bounded (batch) PCollection.
connection_configuration Row : Configuration options to set up the JMS connection.
Note: if connection factory class name is set, spin up a persistent expansion service with the provider client JAR on the classpath:
java -cp
Row fields:
connection_factory_class_name string (Optional) : The JMS ConnectionFactory class name.
password string (Optional) : The password to connect to the JMS broker.
server_uri string : The JMS broker URI.
username string (Optional) : The username to connect to the JMS broker.
queue string (Optional) : The JMS queue to read from. Exclusively one of queue or topic must be specified.
topic string (Optional) : The JMS topic to read from. Exclusively one of queue or topic must be specified.
max_num_records int64 (Optional) : The max number of records to receive. Setting this will result in a bounded PCollection.
max_read_time_seconds int64 (Optional) : The maximum time for this source to read messages. Setting this will result in a bounded PCollection.
close_timeout_seconds int64 (Optional) : Close timeout for the JMS connection in seconds.
acknowledge_mode string (Optional) : The JMS acknowledge mode: CLIENT_ACKNOWLEDGE, CLIENT_ACKNOWLEDGE_UNSAFE, or INDIVIDUAL_ACKNOWLEDGE.
individual_acknowledge_mode_code int32 (Optional) : The proprietary integer code for individual message acknowledgment when using INDIVIDUAL_ACKNOWLEDGE.
type: ReadFromIbmMQ
config:
connection_configuration:
connection_factory_class_name: "connection_factory_class_name"
password: "password"
server_uri: "server_uri"
username: "username"
queue: "queue"
topic: "topic"
max_num_records: max_num_records
max_read_time_seconds: max_read_time_seconds
close_timeout_seconds: close_timeout_seconds
acknowledge_mode: "acknowledge_mode"
individual_acknowledge_mode_code: individual_acknowledge_mode_code
Publishes messages to a JMS broker. Expects an input PCollection of rows with a payload (string) or bytes field, each of which is published as one JMS TextMessage.
Works with both bounded (batch) and unbounded (streaming) input PCollections.
connection_configuration Row : Configuration options to set up the JMS connection.
Note: if connection factory class name is set, spin up a persistent expansion service with the provider client JAR on the classpath:
java -cp
Row fields:
connection_factory_class_name string (Optional) : The JMS ConnectionFactory class name.
password string (Optional) : The password to connect to the JMS broker.
server_uri string : The JMS broker URI.
username string (Optional) : The username to connect to the JMS broker.
queue string (Optional) : The JMS queue to write to. Exclusively one of queue or topic must be specified.
topic string (Optional) : The JMS topic to write to. Exclusively one of queue or topic must be specified.
type: WriteToIbmMQ
input: ...
config:
connection_configuration:
connection_factory_class_name: "connection_factory_class_name"
password: "password"
server_uri: "server_uri"
username: "username"
queue: "queue"
topic: "topic"
Reads an Apache Iceberg table.
See also the Apache Iceberg Beam documentation.
table string : The identifier of the Apache Iceberg table. Example: "db.table1".
filter string (Optional) : SQL-like predicate to filter data at scan time.
Example: "id > 5 AND status = 'ACTIVE'". Uses Apache Calcite syntax:
https://calcite.apache.org/docs/reference.html
keep Array[string] (Optional) : A subset of column names to read exclusively. If null or empty,
all columns will be read.
drop Array[string] (Optional) : A subset of column names to exclude from reading. If null or empty,
all columns will be read.
catalog_name string (Optional) : The name of the catalog. Example: "local".
catalog_properties Map[string, string] (Optional) : A map of configuration properties for the Apache Iceberg
catalog.
The required properties depend on the catalog. For more information, see
CatalogUtil in the Apache Iceberg documentation.
config_properties Map[string, string] (Optional) : An optional set of Hadoop configuration properties.
For more information, see CatalogUtil in the Apache Iceberg documentation.
type: ReadFromIceberg
config:
table: "table"
filter: "filter"
keep:
- "keep"
- "keep"
- ...
drop:
- "drop"
- "drop"
- ...
catalog_name: "catalog_name"
catalog_properties:
a: "catalog_properties_value_a"
b: "catalog_properties_value_b"
c: ...
config_properties:
a: "config_properties_value_a"
b: "config_properties_value_b"
c: ...
Writes to an Apache Iceberg table.
See also the Apache Iceberg Beam documentation including the dynamic destinations section for use of the keep, drop, and only parameters.
table string : The identifier of the Apache Iceberg table. Example: "db.table1".
catalog_name string (Optional) : The name of the catalog. Example: "local".
catalog_properties Map[string, string] (Optional) : A map of configuration properties for the Apache Iceberg
catalog.
The required properties depend on the catalog. For more information, see
CatalogUtil in the Apache Iceberg documentation.
config_properties Map[string, string] (Optional) : An optional set of Hadoop configuration properties.
For more information, see CatalogUtil in the Apache Iceberg documentation.
partition_fields Array[string] (Optional) : Fields used to create a partition spec that is applied
when tables are created. For a field 'foo', the available partition
transforms are:
table_properties Map[string, string] (Optional) : Iceberg table properties to be set on the table when it
is created. For more information on table properties, please visit
https://iceberg.apache.org/docs/latest/configuration/#table-properties.
triggering_frequency_seconds int64 (Optional) : For streaming write pipelines, the frequency
at which the sink attempts to produce snapshots, in seconds.
keep Array[string] (Optional) : An optional list of field names to keep when writing to the
destination. Other fields are dropped. Mutually exclusive with drop
and only.
drop Array[string] (Optional) : An optional list of field names to drop before writing to the
destination. Mutually exclusive with keep and only.
only string (Optional) : The name of exactly one field to keep as the top level record when
writing to the destination. All other fields are dropped. This field must
be of row type. Mutually exclusive with drop and keep.
distribution_mode string (Optional) : Defines distribution of write data. Supported
distributions:
autosharding boolean (Optional) : Enables dynamic sharding to automatically adjust the number
of parallel writers based on data volume. It handles data skew by
further sub-dividing partitions into multiple shards to prevent
bottlenecks during high-throughput writes. Only available with 'hash'
distribution mode.
mode string (Optional) : Controls how rows are written. 'append' (default) appends every row
as new data. 'merge-on-read' treats each row as a change (INSERT,
UPDATE_BEFORE, UPDATE_AFTER, or DELETE) applied to the table by primary
key.
sequence_number_column string (Optional) : Merge-on-read only. the required column name
representing the monotonic sequence number used to order a single key's
changes. Default column name is '_commit_snapshot_sequence_number'. This
column will be stripped from the data row before writing to Iceberg.
change_type_column string (Optional) : Merge-on-read only. The optional column name
representing the row's change type (INSERT, UPDATE_BEFORE, UPDATE_AFTER,
or DELETE). This column will be stripped from the data row before writing
to Iceberg.
change_type_map Map[string, string] (Optional) : Merge-on-read only. Optional map from the
change_type_column value to the canonical change type name (see above).
upsert boolean (Optional) : Merge-on-read only. If true, only the after-image of each change
(INSERT/UPDATE_AFTER) is applied as an upsert. UPDATE_BEFORE records
are dropped. Default: false.
equality_columns Array[string] (Optional) : Columns defining row identity (equality-delete fields).
Defaults to the destination table's identifier (primary-key) fields.
Required if the table doesn't exist yet. Currently only supported in
'merge-on-read' mode.
num_shards int64 (Optional) : The number of deterministic primary-key-hash shards per
destination, i.e. the max write parallelism per destination. Defaults to
shards_per_partition int64 (Optional) : Maximum number of shards a single partition's rows
may occupy. Defaults to num_shards. Currently only supported in
'merge-on-read' mode.
allowed_lateness_seconds int64 (Optional) : How long a late record may lag behind the
watermark before it is dropped entirely. Defaults to 21600 (6 hours).
Currently only supported in 'merge-on-read' mode.
sink_id string (Optional) : A stable identifier namespacing the idempotency tokens written to
each commit. Defaults to a unique per-write UUID. Override with a stable
ID for exactly-once commits across relaunches of a particular streaming
write. Currently only relevant in 'merge-on-read' mode.
token_heartbeat_seconds int64 (Optional) : Streaming only. Refresh each idle
destination's commit token every this many seconds. Disabled by default.
Currently only relevant in 'merge-on-read' mode.
snapshot_properties Map[string, string] (Optional) : Extra properties added to every commit's snapshot
summary. Currently only supported in 'merge-on-read' mode.
error_handling Row (Optional) : Where to send records the sink cannot apply (e.g. an
unknown change type) instead of failing the pipeline. Currently only
supported in 'merge-on-read' mode.
Row fields:
string : Name to use for the output error collectionsorter_memory_mb int64 (Optional) : The in-memory buffer size (MB) for the pre-write sort.
Groups larger than this spill to disk. Default: 100MB. Currently only
relevant in 'merge-on-read' mode.
type: WriteToIceberg
input: ...
config:
table: "table"
catalog_name: "catalog_name"
catalog_properties:
a: "catalog_properties_value_a"
b: "catalog_properties_value_b"
c: ...
config_properties:
a: "config_properties_value_a"
b: "config_properties_value_b"
c: ...
partition_fields:
- "partition_fields"
- "partition_fields"
- ...
table_properties:
a: "table_properties_value_a"
b: "table_properties_value_b"
c: ...
triggering_frequency_seconds: triggering_frequency_seconds
keep:
- "keep"
- "keep"
- ...
drop:
- "drop"
- "drop"
- ...
only: "only"
distribution_mode: "distribution_mode"
autosharding: true|false
mode: "mode"
sequence_number_column: "sequence_number_column"
change_type_column: "change_type_column"
change_type_map:
a: "change_type_map_value_a"
b: "change_type_map_value_b"
c: ...
upsert: true|false
equality_columns:
- "equality_columns"
- "equality_columns"
- ...
num_shards: num_shards
shards_per_partition: shards_per_partition
allowed_lateness_seconds: allowed_lateness_seconds
sink_id: "sink_id"
token_heartbeat_seconds: token_heartbeat_seconds
snapshot_properties:
a: "snapshot_properties_value_a"
b: "snapshot_properties_value_b"
c: ...
error_handling:
a: error_handling_value_a
b: error_handling_value_b
c: ...
sorter_memory_mb: sorter_memory_mb
table string : Identifier of the Iceberg table.
catalog_name string (Optional) : Name of the catalog containing the table.
catalog_properties Map[string, string] (Optional) : Properties used to set up the Iceberg catalog.
config_properties Map[string, string] (Optional) : Properties passed to the Hadoop Configuration.
drop Array[string] (Optional) : A subset of column names to exclude from reading. If null or empty, all columns will be read.
filter string (Optional) : SQL-like predicate to filter data at scan time. Example: "id > 5 AND status = 'ACTIVE'". Uses Apache Calcite syntax: https://calcite.apache.org/docs/reference.html
from_snapshot int64 (Optional) : Starts reading from this snapshot ID (inclusive).
from_timestamp int64 (Optional) : Starts reading from the first snapshot (inclusive) that was created after this timestamp (in milliseconds).
keep Array[string] (Optional) : A subset of column names to read exclusively. If null or empty, all columns will be read.
poll_interval_seconds int32 (Optional) : The interval at which to poll for new snapshots. Defaults to 60 seconds.
starting_strategy string (Optional) : The source's starting strategy. Valid options are: "earliest" or "latest". Can be overriden by setting a starting snapshot or timestamp. Defaults to earliest for batch, and latest for streaming.
streaming boolean (Optional) : Enables streaming reads, where source continuously polls for snapshots forever.
to_snapshot int64 (Optional) : Reads up to this snapshot ID (inclusive).
to_timestamp int64 (Optional) : Reads up to the latest snapshot (inclusive) created before this timestamp (in milliseconds).
type: ReadFromIcebergCDC
config:
table: "table"
catalog_name: "catalog_name"
catalog_properties:
a: "catalog_properties_value_a"
b: "catalog_properties_value_b"
c: ...
config_properties:
a: "config_properties_value_a"
b: "config_properties_value_b"
c: ...
drop:
- "drop"
- "drop"
- ...
filter: "filter"
from_snapshot: from_snapshot
from_timestamp: from_timestamp
keep:
- "keep"
- "keep"
- ...
poll_interval_seconds: poll_interval_seconds
starting_strategy: "starting_strategy"
streaming: true|false
to_snapshot: to_snapshot
to_timestamp: to_timestamp
Read from a JDBC source using a SQL query or by directly accessing a single table.
This transform can be used to read from a JDBC source using either a given JDBC driver jar and class name, or by using one of the default packaged drivers given a jdbc_type.
This transform comes packaged with drivers for several popular JDBC distributions. The following distributions can be declared as the jdbc_type: mysql, oracle, postgres, mssql.
For example, reading a MySQL source using a SQL query:
- type: ReadFromJdbc
config:
jdbc_type: mysql
url: "jdbc:mysql://my-host:3306/database"
query: "SELECT * FROM table"
Note: See the following transforms which are built on top of this transform and simplify this logic for several popular JDBC distributions:
If reading from a JDBC source not listed above, or if it is necessary to use a custom driver not packaged with Beam, one must define a JDBC driver and class name.
For example, reading a MySQL source table:
- type: ReadFromJdbc
config:
driver_jars: "path/to/some/jdbc.jar"
driver_class_name: "com.mysql.jdbc.Driver"
url: "jdbc:mysql://my-host:3306/database"
table: "my-table"
Connection properties are properties sent to the Driver used to connect to the JDBC source. For example, to set the character encoding to UTF-8, one could write:
- type: ReadFromJdbc
config:
connectionProperties: "characterEncoding=UTF-8;"
...
All properties should be semi-colon-delimited (e.g. "key1=value1;key2=value2;")
Secret Manager is supported to avoid storing sensitive credentials such as database passwords in plain text. You can configure secret_manager (e.g. GoogleCloudSecretManager) and provide the secret specification string in JSON format to password.
For example, for Google Cloud Secret Manager:
- type: ReadFromJdbc
config:
jdbc_type: mysql
url: "jdbc:mysql://my-host:3306/database"
username: "my-username"
password: "{\"name\": \"my-db-secret\", \"project\": \"my-project\"}"
secret_manager: "GoogleCloudSecretManager"
query: "SELECT * FROM table"
url string : Connection URL for the JDBC source.
connection_init_sql Array[string] (Optional) : Sets the connection init sql statements used by the Driver. Only MySQL and MariaDB support this.
connection_properties string (Optional) : Used to set connection properties passed to the JDBC driver not already defined as standalone parameter (e.g. username and password can be set using parameters above accordingly). Format of the string must be "key1=value1;key2=value2;".
disable_auto_commit boolean (Optional) : Whether to disable auto commit on read. Defaults to true if not provided. The need for this config varies depending on the database platform. Informix requires this to be set to false while Postgres requires this to be set to true.
driver_class_name string (Optional) : Name of a Java Driver class to use to connect to the JDBC source. For example, "com.mysql.jdbc.Driver".
driver_jars string (Optional) : Comma separated path(s) for the JDBC driver jar(s). This can be a local path or GCS (gs://) path.
fetch_size int32 (Optional) : This method is used to override the size of the data that is going to be fetched and loaded in memory per every database call. It should ONLY be used if the default value throws memory errors.
output_parallelization boolean (Optional) : Whether to reshuffle the resulting PCollection so results are distributed to all workers.
password string (Optional) : Password for the JDBC source. Can be specified as a plain password, or as a secret specification in JSON format if used with a secret manager.
query string (Optional) : SQL query used to query the JDBC source.
table string (Optional) : Name of the table to read from.
partition_column string (Optional) : Name of a column of numeric type that will be used for partitioning.
num_partitions int32 (Optional) : The number of partitions
secret_manager string (Optional) : Secret Manager to use for fetching secret values. Available options: 'GoogleCloudSecretManager', 'GoogleCloudHsmGeneratedSecretManager'. If not set, no secret manager is used and the password is treated as a plain password.
type string (Optional) : Type of JDBC source. When specified, an appropriate default Driver will be packaged with the transform. One of mysql, postgres, oracle, or mssql.
username string (Optional) : Username for the JDBC source.
type: ReadFromJdbc
config:
url: "url"
connection_init_sql:
- "connection_init_sql"
- "connection_init_sql"
- ...
connection_properties: "connection_properties"
disable_auto_commit: true|false
driver_class_name: "driver_class_name"
driver_jars: "driver_jars"
fetch_size: fetch_size
output_parallelization: true|false
password: "password"
query: "query"
table: "table"
partition_column: "partition_column"
num_partitions: num_partitions
secret_manager: "secret_manager"
type: "type"
username: "username"
Write to a JDBC sink using a SQL query or by directly accessing a single table.
This transform can be used to write to a JDBC sink using either a given JDBC driver jar and class name, or by using one of the default packaged drivers given a jdbc_type.
This transform comes packaged with drivers for several popular JDBC distributions. The following distributions can be declared as the jdbc_type: mysql, oracle, postgres, mssql.
For example, writing to a MySQL sink using a SQL query:
- type: WriteToJdbc
config:
jdbc_type: mysql
url: "jdbc:mysql://my-host:3306/database"
query: "INSERT INTO table VALUES(?, ?)"
Note: See the following transforms which are built on top of this transform and simplify this logic for several popular JDBC distributions:
If writing to a JDBC sink not listed above, or if it is necessary to use a custom driver not packaged with Beam, one must define a JDBC driver and class name.
For example, writing to a MySQL table:
- type: WriteToJdbc
config:
driver_jars: "path/to/some/jdbc.jar"
driver_class_name: "com.mysql.jdbc.Driver"
url: "jdbc:mysql://my-host:3306/database"
table: "my-table"
Connection properties are properties sent to the Driver used to connect to the JDBC source. For example, to set the character encoding to UTF-8, one could write:
- type: WriteToJdbc
config:
connectionProperties: "characterEncoding=UTF-8;"
...
All properties should be semi-colon-delimited (e.g. "key1=value1;key2=value2;")
Secret Manager is supported to avoid storing sensitive credentials such as database passwords in plain text. You can configure secret_manager (e.g. GoogleCloudSecretManager) and provide the secret specification string in JSON format to password.
For example, for Google Cloud Secret Manager:
- type: WriteToJdbc
config:
jdbc_type: mysql
url: "jdbc:mysql://my-host:3306/database"
username: "my-username"
password: "{\"name\": \"my-db-secret\", \"project\": \"my-project\"}"
secret_manager: "GoogleCloudSecretManager"
query: "INSERT INTO table VALUES(?, ?)"
url string : Connection URL for the JDBC sink.
auto_sharding boolean (Optional) : If true, enables using a dynamically determined number of shards to write.
connection_init_sql Array[string] (Optional) : Sets the connection init sql statements used by the Driver. Only MySQL and MariaDB support this.
connection_properties string (Optional) : Used to set connection properties passed to the JDBC driver not already defined as standalone parameter (e.g. username and password can be set using parameters above accordingly). Format of the string must be "key1=value1;key2=value2;".
driver_class_name string (Optional) : Name of a Java Driver class to use to connect to the JDBC source. For example, "com.mysql.jdbc.Driver".
driver_jars string (Optional) : Comma separated path(s) for the JDBC driver jar(s). This can be a local path or GCS (gs://) path.
password string (Optional) : Password for the JDBC source. Can be specified as a plain password, or as a secret specification in JSON format if used with a secret manager.
table string (Optional) : Name of the table to write to.
batch_size int64 (Optional)
secret_manager string (Optional) : Secret Manager to use for fetching secret values. Available options: 'GoogleCloudSecretManager', 'GoogleCloudHsmGeneratedSecretManager'. If not set, no secret manager is used and the password is treated as a plain password.
type string (Optional) : Type of JDBC source. When specified, an appropriate default Driver will be packaged with the transform. One of mysql, postgres, oracle, or mssql.
username string (Optional) : Username for the JDBC source.
query string (Optional) : SQL query used to insert records into the JDBC sink.
type: WriteToJdbc
input: ...
config:
url: "url"
auto_sharding: true|false
connection_init_sql:
- "connection_init_sql"
- "connection_init_sql"
- ...
connection_properties: "connection_properties"
driver_class_name: "driver_class_name"
driver_jars: "driver_jars"
password: "password"
table: "table"
batch_size: batch_size
secret_manager: "secret_manager"
type: "type"
username: "username"
query: "query"
Reads messages from a JMS broker and outputs each message as a single payload (string) field.
By default the read is unbounded (streaming): it keeps consuming messages from the specified queue or topic until the pipeline is stopped. Setting maxNumRecords and/or maxReadTimeSeconds bounds the read, producing a bounded (batch) PCollection.
connection_configuration Row : Configuration options to set up the JMS connection.
Note: if connection factory class name is set, spin up a persistent expansion service with the provider client JAR on the classpath:
java -cp
Row fields:
connection_factory_class_name string (Optional) : The JMS ConnectionFactory class name.
password string (Optional) : The password to connect to the JMS broker.
server_uri string : The JMS broker URI.
username string (Optional) : The username to connect to the JMS broker.
queue string (Optional) : The JMS queue to read from. Exclusively one of queue or topic must be specified.
topic string (Optional) : The JMS topic to read from. Exclusively one of queue or topic must be specified.
max_num_records int64 (Optional) : The max number of records to receive. Setting this will result in a bounded PCollection.
max_read_time_seconds int64 (Optional) : The maximum time for this source to read messages. Setting this will result in a bounded PCollection.
close_timeout_seconds int64 (Optional) : Close timeout for the JMS connection in seconds.
acknowledge_mode string (Optional) : The JMS acknowledge mode: CLIENT_ACKNOWLEDGE, CLIENT_ACKNOWLEDGE_UNSAFE, or INDIVIDUAL_ACKNOWLEDGE.
individual_acknowledge_mode_code int32 (Optional) : The proprietary integer code for individual message acknowledgment when using INDIVIDUAL_ACKNOWLEDGE.
type: ReadFromJms
config:
connection_configuration:
connection_factory_class_name: "connection_factory_class_name"
password: "password"
server_uri: "server_uri"
username: "username"
queue: "queue"
topic: "topic"
max_num_records: max_num_records
max_read_time_seconds: max_read_time_seconds
close_timeout_seconds: close_timeout_seconds
acknowledge_mode: "acknowledge_mode"
individual_acknowledge_mode_code: individual_acknowledge_mode_code
Publishes messages to a JMS broker. Expects an input PCollection of rows with a payload (string) or bytes field, each of which is published as one JMS TextMessage.
Works with both bounded (batch) and unbounded (streaming) input PCollections.
connection_configuration Row : Configuration options to set up the JMS connection.
Note: if connection factory class name is set, spin up a persistent expansion service with the provider client JAR on the classpath:
java -cp
Row fields:
connection_factory_class_name string (Optional) : The JMS ConnectionFactory class name.
password string (Optional) : The password to connect to the JMS broker.
server_uri string : The JMS broker URI.
username string (Optional) : The username to connect to the JMS broker.
queue string (Optional) : The JMS queue to write to. Exclusively one of queue or topic must be specified.
topic string (Optional) : The JMS topic to write to. Exclusively one of queue or topic must be specified.
type: WriteToJms
input: ...
config:
connection_configuration:
connection_factory_class_name: "connection_factory_class_name"
password: "password"
server_uri: "server_uri"
username: "username"
queue: "queue"
topic: "topic"
A PTransform for reading json values from files into a PCollection.
string : The file path to read from. The path can contain glob
characters such as * and ?.type: ReadFromJson
config:
path: "path"
A PTransform for writing a PCollection as json values to files.
string : The file path to write to. The files written will
begin with this prefix, followed by a shard identifier (see
num_shards) according to the file_naming parameter.type: WriteToJson
input: ...
config:
path: "path"
schema string (Optional) : The schema in which the data is encoded in the Kafka topic. For AVRO data, this is a schema defined with AVRO schema syntax (https://avro.apache.org/docs/1.10.2/spec.html#schemas). For JSON data, this is a schema defined with JSON-schema syntax (https://json-schema.org/). If a URL to Confluent Schema Registry is provided, then this field is ignored, and the schema is fetched from Confluent Schema Registry.
consumer_config Map[string, string] (Optional) : A list of key-value pairs that act as configuration parameters for Kafka consumers. Most of these configurations will not be needed, but if you need to customize your Kafka consumer, you may use this. See a detailed list: https://docs.confluent.io/platform/current/installation/configuration/consumer-configs.html
format string (Optional) : The encoding format for the data stored in Kafka. Valid options are: RAW,STRING,AVRO,JSON,PROTO
topic string
bootstrap_servers string : A list of host/port pairs to use for establishing the initial connection to the Kafka cluster. The client will make use of all servers irrespective of which servers are specified here for bootstrapping—this list only impacts the initial hosts used to discover the full set of servers. This list should be in the form host1:port1,host2:port2,...
confluent_schema_registry_url string (Optional)
confluent_schema_registry_subject string (Optional)
auto_offset_reset_config string (Optional) : What to do when there is no initial offset in Kafka or if the current offset does not exist any more on the server. (1) earliest: automatically reset the offset to the earliest offset. (2) latest: automatically reset the offset to the latest offset (3) none: throw exception to the consumer if no previous offset is found for the consumer’s group
error_handling Row (Optional) : This option specifies whether and where to output unwritable rows.
Row fields:
string : Name to use for the output error collectionfile_descriptor_path string (Optional) : The path to the Protocol Buffer File Descriptor Set file. This file is used for schema definition and message serialization.
message_name string (Optional) : The name of the Protocol Buffer message to be used for schema extraction and data conversion.
max_read_time_seconds int32 (Optional) : Upper bound of how long to read from Kafka.
type: ReadFromKafka
config:
schema: "schema"
consumer_config:
a: "consumer_config_value_a"
b: "consumer_config_value_b"
c: ...
format: "format"
topic: "topic"
bootstrap_servers: "bootstrap_servers"
confluent_schema_registry_url: "confluent_schema_registry_url"
confluent_schema_registry_subject: "confluent_schema_registry_subject"
auto_offset_reset_config: "auto_offset_reset_config"
error_handling:
output: "output"
file_descriptor_path: "file_descriptor_path"
message_name: "message_name"
max_read_time_seconds: max_read_time_seconds
format string : The encoding format for the data stored in Kafka. Valid options are: RAW,JSON,AVRO,PROTO
topic string
bootstrap_servers string : A list of host/port pairs to use for establishing the initial connection to the Kafka cluster. The client will make use of all servers irrespective of which servers are specified here for bootstrapping—this list only impacts the initial hosts used to discover the full set of servers. | Format: host1:port1,host2:port2,...
producer_config_updates Map[string, string] (Optional) : A list of key-value pairs that act as configuration parameters for Kafka producers. Most of these configurations will not be needed, but if you need to customize your Kafka producer, you may use this. See a detailed list: https://docs.confluent.io/platform/current/installation/configuration/producer-configs.html
error_handling Row (Optional) : This option specifies whether and where to output unwritable rows.
Row fields:
string : Name to use for the output error collectionfile_descriptor_path string (Optional) : The path to the Protocol Buffer File Descriptor Set file. This file is used for schema definition and message serialization.
message_name string (Optional) : The name of the Protocol Buffer message to be used for schema extraction and data conversion.
schema string (Optional)
type: WriteToKafka
input: ...
config:
format: "format"
topic: "topic"
bootstrap_servers: "bootstrap_servers"
producer_config_updates:
a: "producer_config_updates_value_a"
b: "producer_config_updates_value_b"
c: ...
error_handling:
output: "output"
file_descriptor_path: "file_descriptor_path"
message_name: "message_name"
schema: "schema"
Reads records from an Amazon Kinesis stream and outputs Beam Rows with data (bytes) plus stream metadata fields.
stream_name string : Kinesis stream name to read from.
aws_access_key string : AWS access key id.
aws_secret_key string : AWS secret access key.
region string : AWS region, for example us-east-1.
service_endpoint string (Optional) : Optional custom Kinesis service endpoint URI.
verify_certificate boolean (Optional) : Whether to verify TLS certificates. Defaults to true. Never disable in production.
max_num_records int64 (Optional) : Maximum number of records to read. When set, the resulting PCollection is bounded.
max_read_time int64 (Optional) : Maximum read time in milliseconds. When set, the resulting PCollection is bounded.
initial_position_in_stream string (Optional) : Where to start reading in the stream: LATEST, TRIM_HORIZON, or AT_TIMESTAMP.
initial_timestamp_in_stream int64 (Optional) : Epoch millis timestamp used when initial_position_in_stream is AT_TIMESTAMP.
request_records_limit int64 (Optional) : Max records returned by a single GetRecords call (1-10000).
up_to_date_threshold int64 (Optional) : Threshold duration in milliseconds after which a shard is considered up to date.
max_capacity_per_shard int64 (Optional) : Maximum number of records to hold in memory per shard.
watermark_policy string (Optional) : Watermark policy: ARRIVAL_TIME or PROCESSING_TIME.
watermark_idle_duration_threshold int64 (Optional) : Idle duration threshold in milliseconds for ARRIVAL_TIME watermark policy.
rate_limit int64 (Optional) : Fixed delay between GetRecords calls in milliseconds.
type: ReadFromKinesis
config:
stream_name: "stream_name"
aws_access_key: "aws_access_key"
aws_secret_key: "aws_secret_key"
region: "region"
service_endpoint: "service_endpoint"
verify_certificate: true|false
max_num_records: max_num_records
max_read_time: max_read_time
initial_position_in_stream: "initial_position_in_stream"
initial_timestamp_in_stream: initial_timestamp_in_stream
request_records_limit: request_records_limit
up_to_date_threshold: up_to_date_threshold
max_capacity_per_shard: max_capacity_per_shard
watermark_policy: "watermark_policy"
watermark_idle_duration_threshold: watermark_idle_duration_threshold
rate_limit: rate_limit
Writes Beam Rows to an Amazon Kinesis stream. Each input row must include a data (bytes) field, or a payload string/bytes field. Optional per-row partition_key overrides the configured default partition key.
stream_name string : Kinesis stream name to write to.
aws_access_key string : AWS access key id.
aws_secret_key string : AWS secret access key.
region string : AWS region, for example us-east-1.
partition_key string : Default partition key used when the input row does not contain a partition_key field.
service_endpoint string (Optional) : Optional custom Kinesis service endpoint URI.
verify_certificate boolean (Optional) : Whether to verify TLS certificates. Defaults to true. Never disable in production.
aggregation_enabled boolean (Optional) : Enable KPL-compatible record aggregation.
aggregation_max_bytes int64 (Optional) : Max aggregated record size in bytes when aggregation is enabled.
aggregation_max_buffered_time int64 (Optional) : Max time in milliseconds to buffer records when aggregation is enabled.
aggregation_shard_refresh_interval int64 (Optional) : Shard map refresh interval in minutes when aggregation is enabled.
type: WriteToKinesis
input: ...
config:
stream_name: "stream_name"
aws_access_key: "aws_access_key"
aws_secret_key: "aws_secret_key"
region: "region"
partition_key: "partition_key"
service_endpoint: "service_endpoint"
verify_certificate: true|false
aggregation_enabled: true|false
aggregation_max_bytes: aggregation_max_bytes
aggregation_max_buffered_time: aggregation_max_buffered_time
aggregation_shard_refresh_interval: aggregation_shard_refresh_interval
Reads data from MongoDB.
The resulting PCollection consists of rows with fields matching the provided schema.
uri string : The connection URI for the MongoDB server.
database string : The MongoDB database to read from.
collection string : The MongoDB collection to read from.
schema string : The schema in which the data is encoded, defined with JSON-schema syntax (https://json-schema.org/).
filter string (Optional) : An optional BSON filter to apply to the read. This should be a valid JSON string.
error_handling Row (Optional) : This option specifies whether and where to output rows that failed to be read.
Row fields:
string : Name to use for the output error collectiontype: ReadFromMongoDB
config:
database: "database"
collection: "collection"
schema: schema
uri: "uri"
filter:
a: filter_value_a
b: filter_value_b
c: ...
error_handling:
output: "output"
Writes data to MongoDB.
uri string : The connection URI for the MongoDB server.
database string : The MongoDB database to write to.
collection string : The MongoDB collection to write to.
batch_size int64 (Optional) : The number of documents to include in each batch write.
error_handling Row (Optional) : This option specifies whether and where to output unwritable rows. Note: Error handling is currently limited to data conversion failures before sending to the MongoDB driver, as the underlying MongoDbIO does not yet support dead-letter queues for write failures.
Row fields:
string : Name to use for the output error collectiontype: WriteToMongoDB
input: ...
config:
database: "database"
collection: "collection"
uri: "uri"
batch_size: batch_size
error_handling:
output: "output"
Read from a MySQL source using a SQL query or by directly accessing a single table.
This is a special case of ReadFromJdbc that includes the necessary MySQL Driver and classes.
An example of using ReadFromMySql with SQL query:
- type: ReadFromMySql
config:
url: "jdbc:mysql://my-host:3306/database"
query: "SELECT * FROM table"
It is also possible to read a table by specifying a table name. For example, the following configuration will perform a read on an entire table:
- type: ReadFromMySql
config:
url: "jdbc:mysql://my-host:3306/database"
table: "my-table"
It might be necessary to use a custom JDBC driver that is not packaged with this transform. If that is the case, see ReadFromJdbc which allows for more custom configuration.
url string : Connection URL for the JDBC source.
connection_init_sql Array[string] (Optional) : Sets the connection init sql statements used by the Driver. Only MySQL and MariaDB support this.
connection_properties string (Optional) : Used to set connection properties passed to the JDBC driver not already defined as standalone parameter (e.g. username and password can be set using parameters above accordingly). Format of the string must be "key1=value1;key2=value2;".
disable_auto_commit boolean (Optional) : Whether to disable auto commit on read. Defaults to true if not provided. The need for this config varies depending on the database platform. Informix requires this to be set to false while Postgres requires this to be set to true.
fetch_size int32 (Optional) : This method is used to override the size of the data that is going to be fetched and loaded in memory per every database call. It should ONLY be used if the default value throws memory errors.
output_parallelization boolean (Optional) : Whether to reshuffle the resulting PCollection so results are distributed to all workers.
password string (Optional) : Password for the JDBC source. Can be specified as a plain password, or as a secret specification in JSON format if used with a secret manager.
query string (Optional) : SQL query used to query the JDBC source.
table string (Optional) : Name of the table to read from.
partition_column string (Optional) : Name of a column of numeric type that will be used for partitioning.
num_partitions int32 (Optional) : The number of partitions
secret_manager string (Optional) : Secret Manager to use for fetching secret values. Available options: 'GoogleCloudSecretManager', 'GoogleCloudHsmGeneratedSecretManager'. If not set, no secret manager is used and the password is treated as a plain password.
username string (Optional) : Username for the JDBC source.
type: ReadFromMySql
config:
url: "url"
connection_init_sql:
- "connection_init_sql"
- "connection_init_sql"
- ...
connection_properties: "connection_properties"
disable_auto_commit: true|false
fetch_size: fetch_size
output_parallelization: true|false
password: "password"
query: "query"
table: "table"
partition_column: "partition_column"
num_partitions: num_partitions
secret_manager: "secret_manager"
username: "username"
Write to a MySQL sink using a SQL query or by directly accessing a single table.
This is a special case of WriteToJdbc that includes the necessary MySQL Driver and classes.
An example of using WriteToMySql with SQL query:
- type: WriteToMySql
config:
url: "jdbc:mysql://my-host:3306/database"
query: "INSERT INTO table VALUES(?, ?)"
It is also possible to read a table by specifying a table name. For example, the following configuration will perform a read on an entire table:
- type: WriteToMySql
config:
url: "jdbc:mysql://my-host:3306/database"
table: "my-table"
It might be necessary to use a custom JDBC driver that is not packaged with this transform. If that is the case, see WriteToJdbc which allows for more custom configuration.
url string : Connection URL for the JDBC sink.
auto_sharding boolean (Optional) : If true, enables using a dynamically determined number of shards to write.
connection_init_sql Array[string] (Optional) : Sets the connection init sql statements used by the Driver. Only MySQL and MariaDB support this.
connection_properties string (Optional) : Used to set connection properties passed to the JDBC driver not already defined as standalone parameter (e.g. username and password can be set using parameters above accordingly). Format of the string must be "key1=value1;key2=value2;".
password string (Optional) : Password for the JDBC source. Can be specified as a plain password, or as a secret specification in JSON format if used with a secret manager.
table string (Optional) : Name of the table to write to.
batch_size int64 (Optional)
secret_manager string (Optional) : Secret Manager to use for fetching secret values. Available options: 'GoogleCloudSecretManager', 'GoogleCloudHsmGeneratedSecretManager'. If not set, no secret manager is used and the password is treated as a plain password.
username string (Optional) : Username for the JDBC source.
query string (Optional) : SQL query used to insert records into the JDBC sink.
type: WriteToMySql
input: ...
config:
url: "url"
auto_sharding: true|false
connection_init_sql:
- "connection_init_sql"
- "connection_init_sql"
- ...
connection_properties: "connection_properties"
password: "password"
table: "table"
batch_size: batch_size
secret_manager: "secret_manager"
username: "username"
query: "query"
Read from a Oracle source using a SQL query or by directly accessing a single table.
This is a special case of ReadFromJdbc that includes the necessary Oracle Driver and classes.
An example of using ReadFromOracle with SQL query:
- type: ReadFromOracle
config:
url: "jdbc:oracle://my-host:1521/database"
query: "SELECT * FROM table"
It is also possible to read a table by specifying a table name. For example, the following configuration will perform a read on an entire table:
- type: ReadFromOracle
config:
url: "jdbc:oracle://my-host:1521/database"
table: "my-table"
It might be necessary to use a custom JDBC driver that is not packaged with this transform. If that is the case, see ReadFromJdbc which allows for more custom configuration.
url string : Connection URL for the JDBC source.
connection_properties string (Optional) : Used to set connection properties passed to the JDBC driver not already defined as standalone parameter (e.g. username and password can be set using parameters above accordingly). Format of the string must be "key1=value1;key2=value2;".
disable_auto_commit boolean (Optional) : Whether to disable auto commit on read. Defaults to true if not provided. The need for this config varies depending on the database platform. Informix requires this to be set to false while Postgres requires this to be set to true.
fetch_size int32 (Optional) : This method is used to override the size of the data that is going to be fetched and loaded in memory per every database call. It should ONLY be used if the default value throws memory errors.
output_parallelization boolean (Optional) : Whether to reshuffle the resulting PCollection so results are distributed to all workers.
password string (Optional) : Password for the JDBC source. Can be specified as a plain password, or as a secret specification in JSON format if used with a secret manager.
query string (Optional) : SQL query used to query the JDBC source.
table string (Optional) : Name of the table to read from.
partition_column string (Optional) : Name of a column of numeric type that will be used for partitioning.
num_partitions int32 (Optional) : The number of partitions
secret_manager string (Optional) : Secret Manager to use for fetching secret values. Available options: 'GoogleCloudSecretManager', 'GoogleCloudHsmGeneratedSecretManager'. If not set, no secret manager is used and the password is treated as a plain password.
username string (Optional) : Username for the JDBC source.
type: ReadFromOracle
config:
url: "url"
connection_properties: "connection_properties"
disable_auto_commit: true|false
fetch_size: fetch_size
output_parallelization: true|false
password: "password"
query: "query"
table: "table"
partition_column: "partition_column"
num_partitions: num_partitions
secret_manager: "secret_manager"
username: "username"
Write to a Oracle sink using a SQL query or by directly accessing a single table.
This is a special case of WriteToJdbc that includes the necessary Oracle Driver and classes.
An example of using WriteToOracle with SQL query:
- type: WriteToOracle
config:
url: "jdbc:oracle://my-host:1521/database"
query: "INSERT INTO table VALUES(?, ?)"
It is also possible to read a table by specifying a table name. For example, the following configuration will perform a read on an entire table:
- type: WriteToOracle
config:
url: "jdbc:oracle://my-host:1521/database"
table: "my-table"
It might be necessary to use a custom JDBC driver that is not packaged with this transform. If that is the case, see WriteToJdbc which allows for more custom configuration.
url string : Connection URL for the JDBC sink.
auto_sharding boolean (Optional) : If true, enables using a dynamically determined number of shards to write.
connection_properties string (Optional) : Used to set connection properties passed to the JDBC driver not already defined as standalone parameter (e.g. username and password can be set using parameters above accordingly). Format of the string must be "key1=value1;key2=value2;".
password string (Optional) : Password for the JDBC source. Can be specified as a plain password, or as a secret specification in JSON format if used with a secret manager.
table string (Optional) : Name of the table to write to.
batch_size int64 (Optional)
secret_manager string (Optional) : Secret Manager to use for fetching secret values. Available options: 'GoogleCloudSecretManager', 'GoogleCloudHsmGeneratedSecretManager'. If not set, no secret manager is used and the password is treated as a plain password.
username string (Optional) : Username for the JDBC source.
query string (Optional) : SQL query used to insert records into the JDBC sink.
type: WriteToOracle
input: ...
config:
url: "url"
auto_sharding: true|false
connection_properties: "connection_properties"
password: "password"
table: "table"
batch_size: batch_size
secret_manager: "secret_manager"
username: "username"
query: "query"
A PTransform for reading Parquet files.
? (Optional)type: ReadFromParquet
config:
path: path
A PTransform for writing parquet files.
? (Optional)type: WriteToParquet
input: ...
config:
path: path
Read from a Postgres source using a SQL query or by directly accessing a single table.
This is a special case of ReadFromJdbc that includes the necessary Postgres Driver and classes.
An example of using ReadFromPostgres with SQL query:
- type: ReadFromPostgres
config:
url: "jdbc:postgresql://my-host:5432/database"
query: "SELECT * FROM table"
It is also possible to read a table by specifying a table name. For example, the following configuration will perform a read on an entire table:
- type: ReadFromPostgres
config:
url: "jdbc:postgresql://my-host:5432/database"
table: "my-table"
It might be necessary to use a custom JDBC driver that is not packaged with this transform. If that is the case, see ReadFromJdbc which allows for more custom configuration.
url string : Connection URL for the JDBC source.
connection_properties string (Optional) : Used to set connection properties passed to the JDBC driver not already defined as standalone parameter (e.g. username and password can be set using parameters above accordingly). Format of the string must be "key1=value1;key2=value2;".
disable_auto_commit boolean (Optional) : Whether to disable auto commit on read. Defaults to true if not provided. The need for this config varies depending on the database platform. Informix requires this to be set to false while Postgres requires this to be set to true.
fetch_size int32 (Optional) : This method is used to override the size of the data that is going to be fetched and loaded in memory per every database call. It should ONLY be used if the default value throws memory errors.
output_parallelization boolean (Optional) : Whether to reshuffle the resulting PCollection so results are distributed to all workers.
password string (Optional) : Password for the JDBC source. Can be specified as a plain password, or as a secret specification in JSON format if used with a secret manager.
query string (Optional) : SQL query used to query the JDBC source.
table string (Optional) : Name of the table to read from.
partition_column string (Optional) : Name of a column of numeric type that will be used for partitioning.
num_partitions int32 (Optional) : The number of partitions
secret_manager string (Optional) : Secret Manager to use for fetching secret values. Available options: 'GoogleCloudSecretManager', 'GoogleCloudHsmGeneratedSecretManager'. If not set, no secret manager is used and the password is treated as a plain password.
username string (Optional) : Username for the JDBC source.
type: ReadFromPostgres
config:
url: "url"
connection_properties: "connection_properties"
disable_auto_commit: true|false
fetch_size: fetch_size
output_parallelization: true|false
password: "password"
query: "query"
table: "table"
partition_column: "partition_column"
num_partitions: num_partitions
secret_manager: "secret_manager"
username: "username"
Write to a Postgres sink using a SQL query or by directly accessing a single table.
This is a special case of WriteToJdbc that includes the necessary Postgres Driver and classes.
An example of using WriteToPostgres with SQL query:
- type: WriteToPostgres
config:
url: "jdbc:postgresql://my-host:5432/database"
query: "INSERT INTO table VALUES(?, ?)"
It is also possible to read a table by specifying a table name. For example, the following configuration will perform a read on an entire table:
- type: WriteToPostgres
config:
url: "jdbc:postgresql://my-host:5432/database"
table: "my-table"
It might be necessary to use a custom JDBC driver that is not packaged with this transform. If that is the case, see WriteToJdbc which allows for more custom configuration.
url string : Connection URL for the JDBC sink.
auto_sharding boolean (Optional) : If true, enables using a dynamically determined number of shards to write.
connection_properties string (Optional) : Used to set connection properties passed to the JDBC driver not already defined as standalone parameter (e.g. username and password can be set using parameters above accordingly). Format of the string must be "key1=value1;key2=value2;".
password string (Optional) : Password for the JDBC source. Can be specified as a plain password, or as a secret specification in JSON format if used with a secret manager.
table string (Optional) : Name of the table to write to.
batch_size int64 (Optional)
secret_manager string (Optional) : Secret Manager to use for fetching secret values. Available options: 'GoogleCloudSecretManager', 'GoogleCloudHsmGeneratedSecretManager'. If not set, no secret manager is used and the password is treated as a plain password.
username string (Optional) : Username for the JDBC source.
query string (Optional) : SQL query used to insert records into the JDBC sink.
type: WriteToPostgres
input: ...
config:
url: "url"
auto_sharding: true|false
connection_properties: "connection_properties"
password: "password"
table: "table"
batch_size: batch_size
secret_manager: "secret_manager"
username: "username"
query: "query"
Reads messages from Cloud Pub/Sub.
topic string (Optional) : Cloud Pub/Sub topic in the form
"projects/
subscription string (Optional) : Existing Cloud Pub/Sub subscription to use in the
form "projects/
format string : The expected format of the message payload. Currently suported
formats are
payload field whose contents
are the raw bytes of the pubsub message.schema ? (Optional) : Schema specification for the given format.
attributes Array[string] (Optional) : List of attribute keys whose values will be flattened into the
output message as additional fields. For example, if the format is raw
and attributes is ["a", "b"] then this read will produce elements of
the form Row(payload=..., a=..., b=...).
attributes_map string (Optional)
id_attribute string (Optional) : The attribute on incoming Pub/Sub messages to use as a unique
record identifier. When specified, the value of this attribute (which
can be any string that uniquely identifies the record) will be used for
deduplication of messages. If not provided, we cannot guarantee
that no duplicate data will be delivered on the Pub/Sub stream. In this
case, deduplication of the stream will be strictly best effort.
timestamp_attribute string (Optional) : Message value to use as element timestamp. If None,
uses message publishing time as the timestamp.
Timestamp values should be in one of two formats:
2015-10-29T23:41:41.123Z. The sub-second component of the
timestamp is optional, and digits beyond the first three (i.e., time
units smaller than milliseconds) may be ignored.publish_time_field string (Optional) : Field to add to output messages with the Pub/Sub
message publish time. If None, no such field is added.
error_handling Row (Optional) : This option specifies whether and where to output error rows.
Row fields:
string : Name to use for the output error collectiontype: 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"
publish_time_field: "publish_time_field"
error_handling:
output: "output"
Writes messages to Cloud Pub/Sub.
topic string : Cloud Pub/Sub topic in the form "/topics/
format string : How to format the message payload. Currently supported
formats are
schema ? (Optional) : Schema specification for the given format.
attributes Array[string] (Optional) : List of attribute keys whose values will be pulled out as
PubSub message attributes. For example, if the format is raw
and attributes is ["a", "b"] then elements of the form
Row(any_field=..., a=..., b=...) will result in PubSub messages whose
payload has the contents of any_field and whose attribute will be
populated with the values of a and b.
attributes_map string (Optional)
id_attribute string (Optional) : If set, will set an attribute for each Cloud Pub/Sub message
with the given name and a unique value. This attribute can then be used
in a ReadFromPubSub PTransform to deduplicate messages.
timestamp_attribute string (Optional) : If set, will set an attribute for each Cloud Pub/Sub
message with the given name and the message's publish time as the value.
error_handling Row (Optional) : This option specifies whether and where to output error rows.
Row fields:
string : Name to use for the output error collectiontype: 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"
Reads rows from a Snowflake table or query using staged CSV files.
server_name string : Snowflake server name.
username string (Optional) : Snowflake username. Required for password and private key authentication.
password string (Optional) : Snowflake password. Mutually exclusive with OAuth token and private key.
oauth_token string (Optional) : Snowflake OAuth token. Mutually exclusive with password and private key.
private_key string (Optional) : Raw Snowflake private key. Mutually exclusive with password and OAuth token.
private_key_passphrase string (Optional) : Passphrase for the Snowflake private key.
database string : Snowflake database name.
snowflake_schema string : Snowflake schema name.
warehouse string (Optional) : Snowflake warehouse name.
role string (Optional) : Snowflake role.
table string (Optional) : Snowflake table to read from.
query string (Optional) : Snowflake query to read from.
staging_bucket_name string : GCS path used to stage CSV files. The path must end with '/'.
storage_integration_name string : Snowflake storage integration name.
schema string : Output schema encoded using JSON Schema syntax.
quotation_mark string (Optional) : Quotation mark used when parsing staged CSV files.
type: ReadFromSnowflake
config:
server_name: "server_name"
username: "username"
password: "password"
oauth_token: "oauth_token"
private_key: "private_key"
private_key_passphrase: "private_key_passphrase"
database: "database"
snowflake_schema: "snowflake_schema"
warehouse: "warehouse"
role: "role"
table: "table"
query: "query"
staging_bucket_name: "staging_bucket_name"
storage_integration_name: "storage_integration_name"
schema: "schema"
quotation_mark: "quotation_mark"
Writes Beam Rows to Snowflake using batch COPY or streaming Snowpipe.
server_name string : Snowflake server name.
username string (Optional) : Snowflake username. Required for password and private key authentication.
password string (Optional) : Snowflake password. Mutually exclusive with OAuth token and private key.
oauth_token string (Optional) : Snowflake OAuth token. Mutually exclusive with password and private key.
private_key string (Optional) : Raw Snowflake private key. Mutually exclusive with password and OAuth token.
private_key_passphrase string (Optional) : Passphrase for the Snowflake private key.
database string : Snowflake database name.
schema string : Snowflake schema name.
warehouse string (Optional) : Snowflake warehouse name.
role string (Optional) : Snowflake role.
table string (Optional) : Destination Snowflake table. Required for batch writes.
snow_pipe string (Optional) : Snowflake Snowpipe name. Required for streaming writes.
staging_bucket_name string : GCS path used to stage CSV files. The path must end with '/'.
storage_integration_name string : Snowflake storage integration name.
create_disposition string (Optional) : Table creation behavior for batch writes. Supported values are CREATE_IF_NEEDED and CREATE_NEVER.
write_disposition string (Optional) : Write behavior for batch writes. Supported values are APPEND, TRUNCATE, and EMPTY.
quotation_mark string (Optional) : Quotation mark used when writing values to staged CSV files.
flush_row_limit int32 (Optional) : Maximum number of rows to stage before flushing in streaming mode.
flush_time_limit_millis int64 (Optional) : Maximum time in milliseconds before flushing staged rows in streaming mode.
shards_number int32 (Optional) : Number of output shards used when staging files.
debug_mode string (Optional) : Streaming log level. Supported values are ERROR and INFO.
type: WriteToSnowflake
input: ...
config:
server_name: "server_name"
username: "username"
password: "password"
oauth_token: "oauth_token"
private_key: "private_key"
private_key_passphrase: "private_key_passphrase"
database: "database"
schema: "schema"
warehouse: "warehouse"
role: "role"
table: "table"
snow_pipe: "snow_pipe"
staging_bucket_name: "staging_bucket_name"
storage_integration_name: "storage_integration_name"
create_disposition: "create_disposition"
write_disposition: "write_disposition"
quotation_mark: "quotation_mark"
flush_row_limit: flush_row_limit
flush_time_limit_millis: flush_time_limit_millis
shards_number: shards_number
debug_mode: "debug_mode"
Performs a Bulk read from Google Cloud Spanner using a specified SQL query or by directly accessing a single table and its columns.
Both Query and Read APIs are supported. See more information about reading from Cloud Spanner.
Example configuration for performing a read using a SQL query:
- type: ReadFromSpanner
config:
instance_id: 'my-instance-id'
database_id: 'my-database'
query: 'SELECT * FROM table'
It is also possible to read a table by specifying a table name and a list of columns. For example, the following configuration will perform a read on an entire table:
- type: ReadFromSpanner
config:
instance_id: 'my-instance-id'
database_id: 'my-database'
table: 'my-table'
columns: ['col1', 'col2']
Additionally, to read using a Secondary Index, specify the index name:
- type: ReadFromSpanner
config:
instance_id: 'my-instance-id'
database_id: 'my-database'
table: 'my-table'
index: 'my-index'
columns: ['col1', 'col2']
Reads by default use the PartitionQuery API which enforces some limitations on the type of queries that can be used so that the data can be read in parallel. If the query is not supported by the PartitionQuery API, then you can specify a non-partitioned read by setting batching to false.
For example:
- type: ReadFromSpanner
config:
batching: false
...
Note: See SpannerIO for more advanced information.
project string (Optional) : Specifies the GCP project ID.
instance string : Specifies the Cloud Spanner instance.
database string : Specifies the Cloud Spanner database.
table string (Optional) : Specifies the Cloud Spanner table.
query string (Optional) : Specifies the SQL query to execute.
columns Array[string] (Optional) : Specifies the columns to read from the table. This parameter is required when table is specified.
index string (Optional) : Specifies the Index to read from. This parameter can only be specified when using table.
batching boolean (Optional) : Set to false to disable batching. Useful when using a query that is not compatible with the PartitionQuery API. Defaults to true.
error_handling Row (Optional) : This option specifies whether and where to output unwritable rows.
Row fields:
string : Name to use for the output error collectiontype: ReadFromSpanner
config:
project: "project"
instance: "instance"
database: "database"
table: "table"
query: "query"
columns:
- "columns"
- "columns"
- ...
index: "index"
batching: true|false
error_handling:
output: "output"
Performs a bulk write to a Google Cloud Spanner table.
Example configuration for performing a write to a single table:
- type: ReadFromSpanner
config:
project_id: 'my-project-id'
instance_id: 'my-instance-id'
database_id: 'my-database'
table: 'my-table'
Note: See SpannerIO for more advanced information.
project string (Optional) : Specifies the GCP project.
instance string : Specifies the Cloud Spanner instance.
database string : Specifies the Cloud Spanner database.
table string : Specifies the Cloud Spanner table.
error_handling Row (Optional) : Whether and how to handle write errors.
Row fields:
string : Name to use for the output error collectiontype: WriteToSpanner
input: ...
config:
project: "project"
instance: "instance"
database: "database"
table: "table"
error_handling:
output: "output"
project string : Specifies the Cloud Spanner project.
instance string : Specifies the Cloud Spanner instance.
database string : Specifies the Cloud Spanner database.
table string : Specifies the Cloud Spanner table.
change_stream string : Specifies the change stream name.
start_at string : Specifies the time that the change stream should be read from.
end_at string (Optional) : Specifies the end time of the change stream.
type: ReadFromSpannerCDC
config:
project: "project"
instance: "instance"
database: "database"
table: "table"
change_stream: "change_stream"
start_at: "start_at"
end_at: "end_at"
Read from a SQL Server source using a SQL query or by directly accessing a single table.
This is a special case of ReadFromJdbc that includes the necessary SQL Server Driver and classes.
An example of using ReadFromSqlServer with SQL query:
- type: ReadFromSqlServer
config:
url: "jdbc:sqlserver://my-host:1433/database"
query: "SELECT * FROM table"
It is also possible to read a table by specifying a table name. For example, the following configuration will perform a read on an entire table:
- type: ReadFromSqlServer
config:
url: "jdbc:sqlserver://my-host:1433/database"
table: "my-table"
It might be necessary to use a custom JDBC driver that is not packaged with this transform. If that is the case, see ReadFromJdbc which allows for more custom configuration.
url string : Connection URL for the JDBC source.
connection_properties string (Optional) : Used to set connection properties passed to the JDBC driver not already defined as standalone parameter (e.g. username and password can be set using parameters above accordingly). Format of the string must be "key1=value1;key2=value2;".
disable_auto_commit boolean (Optional) : Whether to disable auto commit on read. Defaults to true if not provided. The need for this config varies depending on the database platform. Informix requires this to be set to false while Postgres requires this to be set to true.
fetch_size int32 (Optional) : This method is used to override the size of the data that is going to be fetched and loaded in memory per every database call. It should ONLY be used if the default value throws memory errors.
output_parallelization boolean (Optional) : Whether to reshuffle the resulting PCollection so results are distributed to all workers.
password string (Optional) : Password for the JDBC source. Can be specified as a plain password, or as a secret specification in JSON format if used with a secret manager.
query string (Optional) : SQL query used to query the JDBC source.
table string (Optional) : Name of the table to read from.
partition_column string (Optional) : Name of a column of numeric type that will be used for partitioning.
num_partitions int32 (Optional) : The number of partitions
secret_manager string (Optional) : Secret Manager to use for fetching secret values. Available options: 'GoogleCloudSecretManager', 'GoogleCloudHsmGeneratedSecretManager'. If not set, no secret manager is used and the password is treated as a plain password.
username string (Optional) : Username for the JDBC source.
type: ReadFromSqlServer
config:
url: "url"
connection_properties: "connection_properties"
disable_auto_commit: true|false
fetch_size: fetch_size
output_parallelization: true|false
password: "password"
query: "query"
table: "table"
partition_column: "partition_column"
num_partitions: num_partitions
secret_manager: "secret_manager"
username: "username"
Write to a SQL Server sink using a SQL query or by directly accessing a single table.
This is a special case of WriteToJdbc that includes the necessary SQL Server Driver and classes.
An example of using WriteToSqlServer with SQL query:
- type: WriteToSqlServer
config:
url: "jdbc:sqlserver://my-host:1433/database"
query: "INSERT INTO table VALUES(?, ?)"
It is also possible to read a table by specifying a table name. For example, the following configuration will perform a read on an entire table:
- type: WriteToSqlServer
config:
url: "jdbc:sqlserver://my-host:1433/database"
table: "my-table"
It might be necessary to use a custom JDBC driver that is not packaged with this transform. If that is the case, see WriteToJdbc which allows for more custom configuration.
url string : Connection URL for the JDBC sink.
auto_sharding boolean (Optional) : If true, enables using a dynamically determined number of shards to write.
connection_properties string (Optional) : Used to set connection properties passed to the JDBC driver not already defined as standalone parameter (e.g. username and password can be set using parameters above accordingly). Format of the string must be "key1=value1;key2=value2;".
password string (Optional) : Password for the JDBC source. Can be specified as a plain password, or as a secret specification in JSON format if used with a secret manager.
table string (Optional) : Name of the table to write to.
batch_size int64 (Optional)
secret_manager string (Optional) : Secret Manager to use for fetching secret values. Available options: 'GoogleCloudSecretManager', 'GoogleCloudHsmGeneratedSecretManager'. If not set, no secret manager is used and the password is treated as a plain password.
username string (Optional) : Username for the JDBC source.
query string (Optional) : SQL query used to insert records into the JDBC sink.
type: WriteToSqlServer
input: ...
config:
url: "url"
auto_sharding: true|false
connection_properties: "connection_properties"
password: "password"
table: "table"
batch_size: batch_size
secret_manager: "secret_manager"
username: "username"
query: "query"
Reads data from TFRecord.
file_pattern string : A file glob pattern to read TFRecords from.
coder ? (Optional) : Coder used to decode each record.
compression_type string : Used to handle compressed input files.
Default value is CompressionTypes.AUTO, in which case the file_path's
extension will be used to detect the compression.
validate boolean (Optional) : Boolean flag to verify that the files exist during the
pipeline creation time.
type: ReadFromTFRecord
config:
file_pattern: "file_pattern"
coder: coder
compression_type: "compression_type"
validate: true|false
Writes data to TFRecord.
file_path_prefix string : The file path to write to. The files written will begin
with this prefix, followed by a shard identifier (see num_shards), and
end in a common extension, if given by file_name_suffix.
coder ? (Optional) : Coder used to encode each record.
file_name_suffix string (Optional) : Suffix for the files written.
num_shards int64 (Optional) : The number of files (shards) used for output. If not set, the
default value will be used.
shard_name_template string (Optional) : A template string containing placeholders for
the shard number and shard count. When constructing a filename for a
particular shard number, the upper-case letters 'S' and 'N' are
replaced with the 0-padded shard number and shard count respectively.
This argument can be '' in which case it behaves as if num_shards was
set to 1 and only one file will be generated. The default pattern used
is '-SSSSS-of-NNNNN' if None is passed as the shard_name_template.
compression_type string : Used to handle compressed output files. Typical value
is CompressionTypes.AUTO, in which case the file_path's extension will
be used to detect the compression.
type: WriteToTFRecord
input: ...
config:
file_path_prefix: "file_path_prefix"
coder: coder
file_name_suffix: "file_name_suffix"
num_shards: num_shards
shard_name_template: "shard_name_template"
compression_type: "compression_type"
Reads lines from a text files.
The resulting PCollection consists of rows with a single string field named "line."
string : The file path to read from. The path can contain glob
characters such as * and ?.type: ReadFromText
config:
path: "path"
Writes a PCollection to a (set of) text files(s).
The input must be a PCollection whose schema has exactly one field.
string : The file path to write to. The files written will
begin with this prefix, followed by a shard identifier.type: WriteToText
input: ...
config:
path: "path"