tecton.declarative.KinesisConfig

class tecton.declarative.KinesisConfig(stream_name, region, post_processor, timestamp_field, initial_stream_position, watermark_delay_threshold=datetime.timedelta(days=1), deduplication_columns=None, options=None)

Configuration used to reference a Kinesis stream.

The KinesisConfig class is used to create a reference to an AWS Kinesis stream.

This class used as an input to a StreamSource’s parameter stream_config. This class is not a Tecton Object: it is a grouping of parameters. Declaring this class alone will not register a data source. Instead, declare as part of StreamSource that takes this configuration class instance as a parameter.

Methods

__init__

Instantiates a new KinesisConfig.

__init__(stream_name, region, post_processor, timestamp_field, initial_stream_position, watermark_delay_threshold=datetime.timedelta(days=1), deduplication_columns=None, options=None)

Instantiates a new KinesisConfig.

Parameters
  • stream_name (str) – Name of the Kinesis stream.

  • region (str) – AWS region of the stream, e.g: “us-west-2”.

  • post_processor – Python user defined function f(DataFrame) -> DataFrame that takes in raw Pyspark data source DataFrame and translates it to the DataFrame to be consumed by the Feature View. See an example of post_processor in the User Guide.

  • timestamp_field (str) – (Optional) The timestamp column in this data source that should be used by FilteredSource to filter data from this source, before any feature view transformations are applied. Only required if this source is used with FilteredSource.

  • initial_stream_position (str) – Initial position in stream, e.g: “latest” or “trim_horizon”. More information available in Spark Kinesis Documentation.

  • watermark_delay_threshold (timedelta) – (Default: 24h) Watermark time interval, e.g: timedelta(hours=36), used by Spark Structured Streaming to account for late-arriving data. See: https://docs.tecton.ai/v2/overviews/framework/feature_views/stream_feature_view.html#productionizing-a-stream

  • deduplication_columns (Optional[List[str]]) – (Optional) Columns in the stream data that uniquely identify data records. Used for de-duplicating.

  • options (Optional[Dict[str, str]]) – (Optional) A map of additional Spark readStream options

Returns

A KinesisConfig class instance.

Example of a KinesisConfig declaration:

import pyspark
from tecton import KinesisConfig


# Define our deserialization raw stream translator
def raw_data_deserialization(df:pyspark.sql.DataFrame) -> pyspark.sql.DataFrame:
    from pyspark.sql.functions import col, from_json, from_utc_timestamp
    from pyspark.sql.types import StructType, StringType

    payload_schema = (
      StructType()
            .add('amount', StringType(), False)
            .add('isFraud', StringType(), False)
            .add('timestamp', StringType(), False)
    )

    return (
        df.selectExpr('cast (data as STRING) jsonData')
        .select(from_json('jsonData', payload_schema).alias('payload'))
        .select(
            col('payload.amount').cast('long').alias('amount'),
            col('payload.isFraud').cast('long').alias('isFraud'),
            from_utc_timestamp('payload.timestamp', 'UTC').alias('timestamp')
        )
    )
# Declare KinesisConfig instance object that can be used as argument in `StreamSource`
stream_config = KinesisConfig(
                        stream_name='transaction_events',
                        region='us-west-2',
                        initial_stream_position='latest',
                        timestamp_field='timestamp',
                        post_processor=raw_data_deserialization,
                        options={'roleArn': 'arn:aws:iam::472542229217:role/demo-cross-account-kinesis-ro'}
)