tecton.declarative.FileConfig

class tecton.declarative.FileConfig(uri, file_format, convert_to_glue_format=False, timestamp_field=None, timestamp_format=None, post_processor=None, schema_uri=None, schema_override=None, data_delay=datetime.timedelta(0))

Configuration used to reference a file or directory (S3, etc.)

The FileConfig class is used to create a reference to a file or directory of files in S3, HDFS, or DBFS.

The schema of the data source is inferred from the underlying file(s). It can also be modified using the post_processor parameter.

This class is used as an input to a BatchSource’s parameter batch_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 a part of BatchSource that takes this configuration class instance as a parameter.

Methods

__init__

Instantiates a new FileConfig.
__init__(uri, file_format, convert_to_glue_format=False, timestamp_field=None, timestamp_format=None, post_processor=None, schema_uri=None, schema_override=None, data_delay=datetime.timedelta(0))

Instantiates a new FileConfig.

Parameters

  • uri (str) – S3 or HDFS path to file(s).

  • file_format (str) – File format. “json”, “parquet”, or “csv”

  • convert_to_glue_format – Converts all schema column names to lowercase.

  • timestamp_field (Optional[str]) – 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.

  • timestamp_format (Optional[str]) – Format of string-encoded timestamp column (e.g. “yyyy-MM-dd’T’hh:mm:ss.SSS’Z’”). If the timestamp string cannot be parsed with this format, Tecton will fallback and attempt to use the default timestamp parser.

  • post_processor (Optional[Callable]) – 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.

  • schema_uri (Optional[str]) – A file or subpath of “uri” that can be used for fast schema inference. This is useful for speeding up plan computation for highly partitioned data sources containing many files.

  • schema_override (Optional[StructType]) – A pyspark.sql.types.StructType object that will be used as the schema when reading from the file. If omitted, the schema will be inferred automatically.

  • data_delay (timedelta) – By default, incremental materialization jobs run immediately at the end of the batch schedule period. This parameter configures how long they wait after the end of the period before starting, typically to ensure that all data has landed. For example, if a feature view has a batch_schedule of 1 day and one of the data source inputs has data_delay=timedelta(hours=1) set, then incremental materialization jobs will run at 01:00 UTC.

Returns

A FileConfig class instance.

Example of a FileConfig declaration:

from tecton import FileConfig, BatchSource

def convert_temperature(df):
    from pyspark.sql.functions import udf,col
    from pyspark.sql.types import DoubleType

    # Convert the incoming PySpark DataFrame temperature Celsius to Fahrenheit
    udf_convert = udf(lambda x: x * 1.8 + 32.0, DoubleType())
    converted_df = df.withColumn("Fahrenheit", udf_convert(col("Temperature"))).drop("Temperature")
    return converted_df

# declare a FileConfig, which can be used as a parameter to a `BatchSource`
ad_impressions_file_ds = FileConfig(uri="s3://tecton.ai.public/data/ad_impressions_sample.parquet",
                                    file_format="parquet",
                                    timestamp_field="timestamp",
                                    post_processor=convert_temperature)

# This FileConfig can then be included as an parameter a BatchSource declaration.
# For example,
ad_impressions_batch = BatchSource(name="ad_impressions_batch",
                                   batch_config=ad_impressions_file_ds)

Attributes

data_delay