From 6589a2ec50b79fce3cd371a9ea37024bc3bfc5ee Mon Sep 17 00:00:00 2001 From: Felix Wang Date: Thu, 16 Jun 2022 11:51:47 -0700 Subject: [PATCH 1/7] Deprecate `bootstrap_servers` parameter in KafkaSource Signed-off-by: Felix Wang --- protos/feast/core/DataSource.proto | 2 +- sdk/python/feast/data_source.py | 78 +++++++++++-------- .../integration/registration/test_registry.py | 2 +- .../test_stream_feature_view_apply.py | 4 +- sdk/python/tests/unit/test_data_sources.py | 6 +- sdk/python/tests/unit/test_feature_views.py | 10 +-- 6 files changed, 57 insertions(+), 45 deletions(-) diff --git a/protos/feast/core/DataSource.proto b/protos/feast/core/DataSource.proto index e71066ee709..1a4b95b8479 100644 --- a/protos/feast/core/DataSource.proto +++ b/protos/feast/core/DataSource.proto @@ -128,7 +128,7 @@ message DataSource { // Java Protobuf class at the given class path message KafkaOptions { // Comma separated list of Kafka bootstrap servers. Used for feature tables without a defined source host[:port]] - string bootstrap_servers = 1; + string kafka_bootstrap_servers = 1; // Kafka topic to collect feature data from. string topic = 2; diff --git a/sdk/python/feast/data_source.py b/sdk/python/feast/data_source.py index 3bc9d98e62c..454546f515f 100644 --- a/sdk/python/feast/data_source.py +++ b/sdk/python/feast/data_source.py @@ -50,12 +50,12 @@ class KafkaOptions: def __init__( self, - bootstrap_servers: str, + kafka_bootstrap_servers: str, message_format: StreamFormat, topic: str, watermark: Optional[timedelta] = None, ): - self.bootstrap_servers = bootstrap_servers + self.kafka_bootstrap_servers = kafka_bootstrap_servers self.message_format = message_format self.topic = topic self.watermark = watermark or None @@ -79,7 +79,7 @@ def from_proto(cls, kafka_options_proto: DataSourceProto.KafkaOptions): else kafka_options_proto.watermark.ToTimedelta() ) kafka_options = cls( - bootstrap_servers=kafka_options_proto.bootstrap_servers, + kafka_bootstrap_servers=kafka_options_proto.kafka_bootstrap_servers, message_format=StreamFormat.from_proto(kafka_options_proto.message_format), topic=kafka_options_proto.topic, watermark=watermark, @@ -100,7 +100,7 @@ def to_proto(self) -> DataSourceProto.KafkaOptions: watermark_duration.FromTimedelta(self.watermark) kafka_options_proto = DataSourceProto.KafkaOptions( - bootstrap_servers=self.bootstrap_servers, + kafka_bootstrap_servers=self.kafka_bootstrap_servers, message_format=self.message_format.to_proto(), topic=self.topic, watermark=watermark_duration, @@ -364,20 +364,13 @@ def get_table_query_string(self) -> str: class KafkaSource(DataSource): - def validate(self, config: RepoConfig): - pass - - def get_table_column_names_and_types( - self, config: RepoConfig - ) -> Iterable[Tuple[str, str]]: - pass - def __init__( self, *args, name: Optional[str] = None, event_timestamp_column: Optional[str] = "", bootstrap_servers: Optional[str] = None, + kafka_bootstrap_servers: Optional[str] = None, message_format: Optional[StreamFormat] = None, topic: Optional[str] = None, created_timestamp_column: Optional[str] = "", @@ -391,28 +384,30 @@ def __init__( watermark: Optional[timedelta] = None, ): """ - Creates a KafkaSource stream source object. + Creates a KafkaSource object. + Args: - name: str. Name of data source, which should be unique within a project - event_timestamp_column (optional): str. (Deprecated) Event timestamp column used for point in time + name: Name of data source, which should be unique within a project + event_timestamp_column: (Deprecated) Event timestamp column used for point in time joins of feature values. - bootstrap_servers: str. The servers of the kafka broker in the form "localhost:9092". - message_format: StreamFormat. StreamFormat of serialized messages. - topic: str. The name of the topic to read from in the kafka source. - created_timestamp_column (optional): str. Timestamp column indicating when the row + bootstrap_servers: (Deprecated) The servers of the kafka broker in the form "localhost:9092". + kafka_bootstrap_servers: The servers of the kafka broker in the form "localhost:9092". + message_format: StreamFormat of serialized messages. + topic: The name of the topic to read from in the kafka source. + created_timestamp_column (optional): Timestamp column indicating when the row was created, used for deduplicating rows. - field_mapping (optional): dict(str, str). A dictionary mapping of column names in this data + field_mapping (optional): A dictionary mapping of column names in this data source to feature names in a feature table or view. Only used for feature columns, not entity or timestamp columns. - date_partition_column (optional): str. Timestamp column used for partitioning. - description (optional): str. A human-readable description. - tags (optional): dict(str, str). A dictionary of key-value pairs to store arbitrary metadata. - owner (optional): str. The owner of the data source, typically the email of the primary + date_partition_column (optional): Timestamp column used for partitioning. + description (optional): A human-readable description. + tags (optional): A dictionary of key-value pairs to store arbitrary metadata. + owner (optional): The owner of the data source, typically the email of the primary maintainer. - timestamp_field (optional): str. Event timestamp field used for point + timestamp_field (optional): Event timestamp field used for point in time joins of feature values. - batch_source: DataSource. The datasource that acts as a batch source. - watermark: timedelta. The watermark for stream data. Specifically how late stream data can arrive without being discarded. + batch_source: The datasource that acts as a batch source. + watermark: The watermark for stream data. Specifically how late stream data can arrive without being discarded. """ positional_attributes = [ "name", @@ -423,10 +418,19 @@ def __init__( ] _name = name _event_timestamp_column = event_timestamp_column - _bootstrap_servers = bootstrap_servers or "" + _kafka_bootstrap_servers = kafka_bootstrap_servers or bootstrap_servers or "" _message_format = message_format _topic = topic or "" + if bootstrap_servers: + warnings.warn( + ( + "The 'bootstrap_servers' parameter has been deprecated in favor of 'kafka_bootstrap_servers'. " + "Feast 0.25 and onwards will not support the 'bootstrap_servers' parameter." + ), + DeprecationWarning, + ) + if args: warnings.warn( ( @@ -445,7 +449,7 @@ def __init__( if len(args) >= 2: _event_timestamp_column = args[1] if len(args) >= 3: - _bootstrap_servers = args[2] + _kafka_bootstrap_servers = args[2] if len(args) >= 4: _message_format = args[3] if len(args) >= 5: @@ -471,7 +475,7 @@ def __init__( self.batch_source = batch_source self.kafka_options = KafkaOptions( - bootstrap_servers=_bootstrap_servers, + kafka_bootstrap_servers=_kafka_bootstrap_servers, message_format=_message_format, topic=_topic, watermark=watermark, @@ -487,8 +491,8 @@ def __eq__(self, other): return False if ( - self.kafka_options.bootstrap_servers - != other.kafka_options.bootstrap_servers + self.kafka_options.kafka_bootstrap_servers + != other.kafka_options.kafka_bootstrap_servers or self.kafka_options.message_format != other.kafka_options.message_format or self.kafka_options.topic != other.kafka_options.topic or self.kafka_options.watermark != other.kafka_options.watermark @@ -513,7 +517,7 @@ def from_proto(data_source: DataSourceProto): name=data_source.name, event_timestamp_column=data_source.timestamp_field, field_mapping=dict(data_source.field_mapping), - bootstrap_servers=data_source.kafka_options.bootstrap_servers, + kafka_bootstrap_servers=data_source.kafka_options.kafka_bootstrap_servers, message_format=StreamFormat.from_proto( data_source.kafka_options.message_format ), @@ -548,6 +552,14 @@ def to_proto(self) -> DataSourceProto: data_source_proto.batch_source.MergeFrom(self.batch_source.to_proto()) return data_source_proto + def validate(self, config: RepoConfig): + pass + + def get_table_column_names_and_types( + self, config: RepoConfig + ) -> Iterable[Tuple[str, str]]: + pass + @staticmethod def source_datatype_to_feast_value_type() -> Callable[[str], ValueType]: return type_map.redshift_to_feast_value_type diff --git a/sdk/python/tests/integration/registration/test_registry.py b/sdk/python/tests/integration/registration/test_registry.py index fcf65570a06..0a7b4e2bbef 100644 --- a/sdk/python/tests/integration/registration/test_registry.py +++ b/sdk/python/tests/integration/registration/test_registry.py @@ -315,7 +315,7 @@ def simple_udf(x: int): stream_source = KafkaSource( name="kafka", timestamp_field="event_timestamp", - bootstrap_servers="", + kafka_bootstrap_servers="", message_format=AvroFormat(""), topic="topic", batch_source=FileSource(path="some path"), diff --git a/sdk/python/tests/integration/registration/test_stream_feature_view_apply.py b/sdk/python/tests/integration/registration/test_stream_feature_view_apply.py index adeb15317e3..14e718d2305 100644 --- a/sdk/python/tests/integration/registration/test_stream_feature_view_apply.py +++ b/sdk/python/tests/integration/registration/test_stream_feature_view_apply.py @@ -29,7 +29,7 @@ def test_apply_stream_feature_view(simple_dataset_1) -> None: stream_source = KafkaSource( name="kafka", timestamp_field="event_timestamp", - bootstrap_servers="", + kafka_bootstrap_servers="", message_format=AvroFormat(""), topic="topic", batch_source=file_source, @@ -93,7 +93,7 @@ def test_stream_feature_view_udf(simple_dataset_1) -> None: stream_source = KafkaSource( name="kafka", timestamp_field="event_timestamp", - bootstrap_servers="", + kafka_bootstrap_servers="", message_format=AvroFormat(""), topic="topic", batch_source=file_source, diff --git a/sdk/python/tests/unit/test_data_sources.py b/sdk/python/tests/unit/test_data_sources.py index 7f288d36db9..f3749a5c777 100644 --- a/sdk/python/tests/unit/test_data_sources.py +++ b/sdk/python/tests/unit/test_data_sources.py @@ -93,7 +93,7 @@ def test_default_data_source_kw_arg_warning(): ) assert source.name == "name" assert source.timestamp_field == "column" - assert source.kafka_options.bootstrap_servers == "bootstrap_servers" + assert source.kafka_options.kafka_bootstrap_servers == "bootstrap_servers" assert source.kafka_options.topic == "topic" with pytest.raises(ValueError): KafkaSource("name", "column", "bootstrap_servers", topic="topic") @@ -145,7 +145,7 @@ def test_default_data_source_kw_arg_warning(): with pytest.warns(UserWarning): source = KafkaSource( timestamp_field="column", - bootstrap_servers="bootstrap_servers", + kafka_bootstrap_servers="bootstrap_servers", message_format=ProtoFormat("class_path"), topic="topic", ) @@ -203,7 +203,7 @@ def test_proto_conversion(): kafka_source = KafkaSource( name="test_source", - bootstrap_servers="test_servers", + kafka_bootstrap_servers="test_servers", message_format=ProtoFormat("class_path"), topic="test_topic", timestamp_field="event_timestamp", diff --git a/sdk/python/tests/unit/test_feature_views.py b/sdk/python/tests/unit/test_feature_views.py index a1d134a2f0e..849d2d0aed9 100644 --- a/sdk/python/tests/unit/test_feature_views.py +++ b/sdk/python/tests/unit/test_feature_views.py @@ -31,7 +31,7 @@ def test_create_batch_feature_view(): stream_source = KafkaSource( name="kafka", timestamp_field="event_timestamp", - bootstrap_servers="", + kafka_bootstrap_servers="", message_format=AvroFormat(""), topic="topic", batch_source=FileSource(path="some path"), @@ -49,7 +49,7 @@ def test_create_stream_feature_view(): stream_source = KafkaSource( name="kafka", timestamp_field="event_timestamp", - bootstrap_servers="", + kafka_bootstrap_servers="", message_format=AvroFormat(""), topic="topic", batch_source=FileSource(path="some path"), @@ -100,7 +100,7 @@ def test_stream_feature_view_serialization(): stream_source = KafkaSource( name="kafka", timestamp_field="event_timestamp", - bootstrap_servers="", + kafka_bootstrap_servers="", message_format=AvroFormat(""), topic="topic", batch_source=FileSource(path="some path"), @@ -137,7 +137,7 @@ def test_stream_feature_view_udfs(): stream_source = KafkaSource( name="kafka", timestamp_field="event_timestamp", - bootstrap_servers="", + kafka_bootstrap_servers="", message_format=AvroFormat(""), topic="topic", batch_source=FileSource(path="some path"), @@ -183,7 +183,7 @@ def test_stream_feature_view_initialization_with_optional_fields_omitted(): stream_source = KafkaSource( name="kafka", timestamp_field="event_timestamp", - bootstrap_servers="", + kafka_bootstrap_servers="", message_format=AvroFormat(""), topic="topic", batch_source=FileSource(path="some path"), From 25808357b8faac082f4d99ece7bbac2f8400cc78 Mon Sep 17 00:00:00 2001 From: Felix Wang Date: Thu, 16 Jun 2022 12:07:56 -0700 Subject: [PATCH 2/7] Rename `watermark` to `watermark_delay_threshold` for KafkaSource Signed-off-by: Felix Wang --- protos/feast/core/DataSource.proto | 3 +- sdk/python/feast/data_source.py | 49 ++++++++++--------- .../integration/registration/test_registry.py | 2 +- .../test_stream_feature_view_apply.py | 4 +- 4 files changed, 31 insertions(+), 27 deletions(-) diff --git a/protos/feast/core/DataSource.proto b/protos/feast/core/DataSource.proto index 1a4b95b8479..62f5859ee8e 100644 --- a/protos/feast/core/DataSource.proto +++ b/protos/feast/core/DataSource.proto @@ -136,7 +136,8 @@ message DataSource { // Defines the stream data format encoding feature/entity data in Kafka messages. StreamFormat message_format = 3; - google.protobuf.Duration watermark = 4; + // Watermark delay threshold for stream data + google.protobuf.Duration watermark_delay_threshold = 4; } // Defines options for DataSource that sources features from Kinesis records. diff --git a/sdk/python/feast/data_source.py b/sdk/python/feast/data_source.py index 454546f515f..2d7fe23dc26 100644 --- a/sdk/python/feast/data_source.py +++ b/sdk/python/feast/data_source.py @@ -53,12 +53,12 @@ def __init__( kafka_bootstrap_servers: str, message_format: StreamFormat, topic: str, - watermark: Optional[timedelta] = None, + watermark_delay_threshold: Optional[timedelta] = None, ): self.kafka_bootstrap_servers = kafka_bootstrap_servers self.message_format = message_format self.topic = topic - self.watermark = watermark or None + self.watermark_delay_threshold = watermark_delay_threshold or None @classmethod def from_proto(cls, kafka_options_proto: DataSourceProto.KafkaOptions): @@ -71,18 +71,18 @@ def from_proto(cls, kafka_options_proto: DataSourceProto.KafkaOptions): Returns: Returns a BigQueryOptions object based on the kafka_options protobuf """ - watermark = None - if kafka_options_proto.HasField("watermark"): - watermark = ( + watermark_delay_threshold = None + if kafka_options_proto.HasField("watermark_delay_threshold"): + watermark_delay_threshold = ( timedelta(days=0) - if kafka_options_proto.watermark.ToNanoseconds() == 0 - else kafka_options_proto.watermark.ToTimedelta() + if kafka_options_proto.watermark_delay_threshold.ToNanoseconds() == 0 + else kafka_options_proto.watermark_delay_threshold.ToTimedelta() ) kafka_options = cls( kafka_bootstrap_servers=kafka_options_proto.kafka_bootstrap_servers, message_format=StreamFormat.from_proto(kafka_options_proto.message_format), topic=kafka_options_proto.topic, - watermark=watermark, + watermark_delay_threshold=watermark_delay_threshold, ) return kafka_options @@ -94,16 +94,16 @@ def to_proto(self) -> DataSourceProto.KafkaOptions: Returns: KafkaOptionsProto protobuf """ - watermark_duration = None - if self.watermark is not None: - watermark_duration = Duration() - watermark_duration.FromTimedelta(self.watermark) + watermark_delay_threshold = None + if self.watermark_delay_threshold is not None: + watermark_delay_threshold = Duration() + watermark_delay_threshold.FromTimedelta(self.watermark_delay_threshold) kafka_options_proto = DataSourceProto.KafkaOptions( kafka_bootstrap_servers=self.kafka_bootstrap_servers, message_format=self.message_format.to_proto(), topic=self.topic, - watermark=watermark_duration, + watermark_delay_threshold=watermark_delay_threshold, ) return kafka_options_proto @@ -381,7 +381,7 @@ def __init__( owner: Optional[str] = "", timestamp_field: Optional[str] = "", batch_source: Optional[DataSource] = None, - watermark: Optional[timedelta] = None, + watermark_delay_threshold: Optional[timedelta] = None, ): """ Creates a KafkaSource object. @@ -407,7 +407,8 @@ def __init__( timestamp_field (optional): Event timestamp field used for point in time joins of feature values. batch_source: The datasource that acts as a batch source. - watermark: The watermark for stream data. Specifically how late stream data can arrive without being discarded. + watermark_delay_threshold: The watermark delay threshold for stream data. Specifically how + late stream data can arrive without being discarded. """ positional_attributes = [ "name", @@ -478,7 +479,7 @@ def __init__( kafka_bootstrap_servers=_kafka_bootstrap_servers, message_format=_message_format, topic=_topic, - watermark=watermark, + watermark_delay_threshold=watermark_delay_threshold, ) def __eq__(self, other): @@ -495,7 +496,8 @@ def __eq__(self, other): != other.kafka_options.kafka_bootstrap_servers or self.kafka_options.message_format != other.kafka_options.message_format or self.kafka_options.topic != other.kafka_options.topic - or self.kafka_options.watermark != other.kafka_options.watermark + or self.kafka_options.watermark_delay_threshold + != other.kafka_options.watermark_delay_threshold ): return False @@ -506,12 +508,13 @@ def __hash__(self): @staticmethod def from_proto(data_source: DataSourceProto): - watermark = None - if data_source.kafka_options.watermark: - watermark = ( + watermark_delay_threshold = None + if data_source.kafka_options.watermark_delay_threshold: + watermark_delay_threshold = ( timedelta(days=0) - if data_source.kafka_options.watermark.ToNanoseconds() == 0 - else data_source.kafka_options.watermark.ToTimedelta() + if data_source.kafka_options.watermark_delay_threshold.ToNanoseconds() + == 0 + else data_source.kafka_options.watermark_delay_threshold.ToTimedelta() ) return KafkaSource( name=data_source.name, @@ -521,7 +524,7 @@ def from_proto(data_source: DataSourceProto): message_format=StreamFormat.from_proto( data_source.kafka_options.message_format ), - watermark=watermark, + watermark_delay_threshold=watermark_delay_threshold, topic=data_source.kafka_options.topic, created_timestamp_column=data_source.created_timestamp_column, timestamp_field=data_source.timestamp_field, diff --git a/sdk/python/tests/integration/registration/test_registry.py b/sdk/python/tests/integration/registration/test_registry.py index 0a7b4e2bbef..e2ee4ee33ff 100644 --- a/sdk/python/tests/integration/registration/test_registry.py +++ b/sdk/python/tests/integration/registration/test_registry.py @@ -319,7 +319,7 @@ def simple_udf(x: int): message_format=AvroFormat(""), topic="topic", batch_source=FileSource(path="some path"), - watermark=timedelta(days=1), + watermark_delay_threshold=timedelta(days=1), ) sfv = StreamFeatureView( diff --git a/sdk/python/tests/integration/registration/test_stream_feature_view_apply.py b/sdk/python/tests/integration/registration/test_stream_feature_view_apply.py index 14e718d2305..8e2af031c5f 100644 --- a/sdk/python/tests/integration/registration/test_stream_feature_view_apply.py +++ b/sdk/python/tests/integration/registration/test_stream_feature_view_apply.py @@ -33,7 +33,7 @@ def test_apply_stream_feature_view(simple_dataset_1) -> None: message_format=AvroFormat(""), topic="topic", batch_source=file_source, - watermark=timedelta(days=1), + watermark_delay_threshold=timedelta(days=1), ) @stream_feature_view( @@ -97,7 +97,7 @@ def test_stream_feature_view_udf(simple_dataset_1) -> None: message_format=AvroFormat(""), topic="topic", batch_source=file_source, - watermark=timedelta(days=1), + watermark_delay_threshold=timedelta(days=1), ) @stream_feature_view( From 2c970535a230297b9ec3c1cb22234fd9366b4f0c Mon Sep 17 00:00:00 2001 From: Felix Wang Date: Thu, 16 Jun 2022 11:59:48 -0700 Subject: [PATCH 3/7] Deprecate `date_partition_column` for all data sources Signed-off-by: Felix Wang --- sdk/python/feast/data_source.py | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/sdk/python/feast/data_source.py b/sdk/python/feast/data_source.py index 2d7fe23dc26..0b2918b8c8f 100644 --- a/sdk/python/feast/data_source.py +++ b/sdk/python/feast/data_source.py @@ -260,6 +260,14 @@ def __init__( self.date_partition_column = ( date_partition_column if date_partition_column else "" ) + if date_partition_column: + warnings.warn( + ( + "The argument 'date_partition_column' is being deprecated. " + "Feast 0.25 and onwards will not support 'date_timestamp_column' for data sources." + ), + DeprecationWarning, + ) self.description = description or "" self.tags = tags or {} self.owner = owner or "" From aafd5e92ab7e244e482cda84942c8932c9ccef44 Mon Sep 17 00:00:00 2001 From: Felix Wang Date: Thu, 16 Jun 2022 15:17:19 -0700 Subject: [PATCH 4/7] Fix Java Signed-off-by: Felix Wang --- .../serving/src/test/java/feast/serving/util/DataGenerator.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/java/serving/src/test/java/feast/serving/util/DataGenerator.java b/java/serving/src/test/java/feast/serving/util/DataGenerator.java index 7a310828d2e..fd00262763b 100644 --- a/java/serving/src/test/java/feast/serving/util/DataGenerator.java +++ b/java/serving/src/test/java/feast/serving/util/DataGenerator.java @@ -226,7 +226,7 @@ public static DataSource createKafkaDataSourceSpec( .setKafkaOptions( KafkaOptions.newBuilder() .setTopic(topic) - .setBootstrapServers(servers) + .setKafkaBootstrapServers(servers) .setMessageFormat(createProtoFormat("class.path")) .build()) .setTimestampField(timestampColumn) From 5d4b055853146a4abea70b42c020e432b0b81d1a Mon Sep 17 00:00:00 2001 From: Felix Wang Date: Thu, 16 Jun 2022 20:38:15 -0700 Subject: [PATCH 5/7] Fix SparkKafkaProcessor Signed-off-by: Felix Wang --- sdk/python/feast/infra/contrib/spark_kafka_processor.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/sdk/python/feast/infra/contrib/spark_kafka_processor.py b/sdk/python/feast/infra/contrib/spark_kafka_processor.py index 57361e5a18f..4dfb615773c 100644 --- a/sdk/python/feast/infra/contrib/spark_kafka_processor.py +++ b/sdk/python/feast/infra/contrib/spark_kafka_processor.py @@ -77,7 +77,7 @@ def _ingest_stream_data(self) -> StreamTable: self.spark.readStream.format("kafka") .option( "kafka.bootstrap.servers", - self.data_source.kafka_options.bootstrap_servers, + self.data_source.kafka_options.kafka_bootstrap_servers, ) .option("subscribe", self.data_source.kafka_options.topic) .option("startingOffsets", "latest") # Query start @@ -100,7 +100,7 @@ def _ingest_stream_data(self) -> StreamTable: self.spark.readStream.format("kafka") .option( "kafka.bootstrap.servers", - self.data_source.kafka_options.bootstrap_servers, + self.data_source.kafka_options.kafka_bootstrap_servers, ) .option("subscribe", self.data_source.kafka_options.topic) .option("startingOffsets", "latest") # Query start From a882b748f7d8b05bc30ac8513ce72c81c91c2f87 Mon Sep 17 00:00:00 2001 From: Felix Wang Date: Thu, 16 Jun 2022 21:39:12 -0700 Subject: [PATCH 6/7] Clarify comment Signed-off-by: Felix Wang --- sdk/python/feast/data_source.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/sdk/python/feast/data_source.py b/sdk/python/feast/data_source.py index 0b2918b8c8f..1ccc2e003e5 100644 --- a/sdk/python/feast/data_source.py +++ b/sdk/python/feast/data_source.py @@ -178,8 +178,8 @@ class DataSource(ABC): Args: name: Name of data source, which should be unique within a project - event_timestamp_column (optional): (Deprecated) Event timestamp column used for point in time - joins of feature values. + event_timestamp_column (optional): (Deprecated in favor of timestamp_field) Event + timestamp column used for point in time joins of feature values. created_timestamp_column (optional): Timestamp column indicating when the row was created, used for deduplicating rows. field_mapping (optional): A dictionary mapping of column names in this data From fc12f540c651562f7fe3eecd1cd784cbbb63a02f Mon Sep 17 00:00:00 2001 From: Felix Wang Date: Thu, 16 Jun 2022 21:53:18 -0700 Subject: [PATCH 7/7] More clarifications Signed-off-by: Felix Wang --- sdk/python/feast/data_source.py | 8 ++++---- sdk/python/feast/infra/offline_stores/bigquery_source.py | 3 ++- sdk/python/feast/infra/offline_stores/file_source.py | 3 ++- sdk/python/feast/infra/offline_stores/redshift_source.py | 4 ++-- sdk/python/feast/infra/offline_stores/snowflake_source.py | 4 ++-- 5 files changed, 12 insertions(+), 10 deletions(-) diff --git a/sdk/python/feast/data_source.py b/sdk/python/feast/data_source.py index 1ccc2e003e5..5f409edb448 100644 --- a/sdk/python/feast/data_source.py +++ b/sdk/python/feast/data_source.py @@ -220,8 +220,8 @@ def __init__( Creates a DataSource object. Args: name: Name of data source, which should be unique within a project - event_timestamp_column (optional): (Deprecated) Event timestamp column used for point in time - joins of feature values. + event_timestamp_column (optional): (Deprecated in favor of timestamp_field) Event + timestamp column used for point in time joins of feature values. created_timestamp_column (optional): Timestamp column indicating when the row was created, used for deduplicating rows. field_mapping (optional): A dictionary mapping of column names in this data @@ -396,8 +396,8 @@ def __init__( Args: name: Name of data source, which should be unique within a project - event_timestamp_column: (Deprecated) Event timestamp column used for point in time - joins of feature values. + event_timestamp_column (optional): (Deprecated in favor of timestamp_field) Event + timestamp column used for point in time joins of feature values. bootstrap_servers: (Deprecated) The servers of the kafka broker in the form "localhost:9092". kafka_bootstrap_servers: The servers of the kafka broker in the form "localhost:9092". message_format: StreamFormat of serialized messages. diff --git a/sdk/python/feast/infra/offline_stores/bigquery_source.py b/sdk/python/feast/infra/offline_stores/bigquery_source.py index 06e9ce9f624..68247a70be4 100644 --- a/sdk/python/feast/infra/offline_stores/bigquery_source.py +++ b/sdk/python/feast/infra/offline_stores/bigquery_source.py @@ -37,7 +37,8 @@ def __init__( Args: table (optional): The BigQuery table where features can be found. - event_timestamp_column: (Deprecated) Event timestamp column used for point in time joins of feature values. + event_timestamp_column (optional): (Deprecated in favor of timestamp_field) Event + timestamp column used for point in time joins of feature values. created_timestamp_column (optional): Timestamp column when row was created, used for deduplicating rows. field_mapping: A dictionary mapping of column names in this data source to feature names in a feature table or view. Only used for feature columns, not entities or timestamp columns. diff --git a/sdk/python/feast/infra/offline_stores/file_source.py b/sdk/python/feast/infra/offline_stores/file_source.py index 85baa647033..6c2ff708c3e 100644 --- a/sdk/python/feast/infra/offline_stores/file_source.py +++ b/sdk/python/feast/infra/offline_stores/file_source.py @@ -44,7 +44,8 @@ def __init__( path: File path to file containing feature data. Must contain an event_timestamp column, entity columns and feature columns. - event_timestamp_column(optional): (Deprecated) Event timestamp column used for point in time joins of feature values. + event_timestamp_column (optional): (Deprecated in favor of timestamp_field) Event + timestamp column used for point in time joins of feature values. created_timestamp_column (optional): Timestamp column when row was created, used for deduplicating rows. file_format (optional): Explicitly set the file format. Allows Feast to bypass inferring the file format. field_mapping: A dictionary mapping of column names in this data source to feature names in a feature table diff --git a/sdk/python/feast/infra/offline_stores/redshift_source.py b/sdk/python/feast/infra/offline_stores/redshift_source.py index ae9d8bab5c8..c6c717e67f9 100644 --- a/sdk/python/feast/infra/offline_stores/redshift_source.py +++ b/sdk/python/feast/infra/offline_stores/redshift_source.py @@ -39,8 +39,8 @@ def __init__( Creates a RedshiftSource object. Args: - event_timestamp_column (optional): (Deprecated) Event timestamp column used for point in - time joins of feature values. + event_timestamp_column (optional): (Deprecated in favor of timestamp_field) Event + timestamp column used for point in time joins of feature values. table (optional): Redshift table where the features are stored. schema (optional): Redshift schema in which the table is located. created_timestamp_column (optional): Timestamp column indicating when the diff --git a/sdk/python/feast/infra/offline_stores/snowflake_source.py b/sdk/python/feast/infra/offline_stores/snowflake_source.py index d76131f837b..323ea79326e 100644 --- a/sdk/python/feast/infra/offline_stores/snowflake_source.py +++ b/sdk/python/feast/infra/offline_stores/snowflake_source.py @@ -43,8 +43,8 @@ def __init__( warehouse (optional): Snowflake warehouse where the database is stored. schema (optional): Snowflake schema in which the table is located. table (optional): Snowflake table where the features are stored. - event_timestamp_column (optional): (Deprecated) Event timestamp column used for point in - time joins of feature values. + event_timestamp_column (optional): (Deprecated in favor of timestamp_field) Event + timestamp column used for point in time joins of feature values. query (optional): The query to be executed to obtain the features. created_timestamp_column (optional): Timestamp column indicating when the row was created, used for deduplicating rows.