From b67c74865d4d350d07b46b0740990a3ae3e4bd40 Mon Sep 17 00:00:00 2001 From: Oleksii Moskalenko Date: Mon, 9 Nov 2020 17:14:41 +0800 Subject: [PATCH] make created timestamp optional in KafkaSource Signed-off-by: Oleksii Moskalenko --- sdk/python/feast/data_source.py | 2 +- sdk/python/tests/test_client.py | 1 - sdk/python/tests/test_feature_table.py | 1 - .../src/main/scala/feast/ingestion/BasePipeline.scala | 2 +- tests/e2e/test_online_features.py | 7 +++---- tests/e2e/test_register.py | 1 - 6 files changed, 5 insertions(+), 9 deletions(-) diff --git a/sdk/python/feast/data_source.py b/sdk/python/feast/data_source.py index 27fc8924f0f..4d80a114578 100644 --- a/sdk/python/feast/data_source.py +++ b/sdk/python/feast/data_source.py @@ -611,10 +611,10 @@ class KafkaSource(DataSource): def __init__( self, event_timestamp_column: str, - created_timestamp_column: str, bootstrap_servers: str, message_format: StreamFormat, topic: str, + created_timestamp_column: Optional[str] = "", field_mapping: Optional[Dict[str, str]] = dict(), date_partition_column: Optional[str] = "", ): diff --git a/sdk/python/tests/test_client.py b/sdk/python/tests/test_client.py index 188e2542db0..564a7b671cc 100644 --- a/sdk/python/tests/test_client.py +++ b/sdk/python/tests/test_client.py @@ -411,7 +411,6 @@ def test_apply_feature_table_success(self, test_client): message_format=ProtoFormat("class.path"), topic="test_topic", event_timestamp_column="ts_col", - created_timestamp_column="timestamp", ) ft1 = FeatureTable( diff --git a/sdk/python/tests/test_feature_table.py b/sdk/python/tests/test_feature_table.py index 2994569a617..575bc605688 100644 --- a/sdk/python/tests/test_feature_table.py +++ b/sdk/python/tests/test_feature_table.py @@ -78,7 +78,6 @@ def test_feature_table_import_export_yaml(self, batch_source): message_format=ProtoFormat(class_path="class.path"), topic="test_topic", event_timestamp_column="ts_col", - created_timestamp_column="timestamp", ) test_feature_table = FeatureTable( diff --git a/spark/ingestion/src/main/scala/feast/ingestion/BasePipeline.scala b/spark/ingestion/src/main/scala/feast/ingestion/BasePipeline.scala index 3257dba69ea..cc1f451ae74 100644 --- a/spark/ingestion/src/main/scala/feast/ingestion/BasePipeline.scala +++ b/spark/ingestion/src/main/scala/feast/ingestion/BasePipeline.scala @@ -63,7 +63,7 @@ trait BasePipeline { jobConfig.stencilURL match { case Some(url: String) => conf - .set("feast.ingestion.registry.proto.kind", "local") + .set("feast.ingestion.registry.proto.kind", "stencil") .set("feast.ingestion.registry.proto.url", url) case None => () } diff --git a/tests/e2e/test_online_features.py b/tests/e2e/test_online_features.py index ab2e256a85f..9e927d5c59a 100644 --- a/tests/e2e/test_online_features.py +++ b/tests/e2e/test_online_features.py @@ -119,10 +119,9 @@ def test_streaming_ingestion( file_url=os.path.join(local_staging_path, "batch-storage"), ), stream_source=KafkaSource( - "event_timestamp", - "event_timestamp", - kafka_broker, - AvroFormat(avro_schema()), + event_timestamp_column="event_timestamp", + bootstrap_servers=kafka_broker, + message_format=AvroFormat(avro_schema()), topic=topic_name, ), ) diff --git a/tests/e2e/test_register.py b/tests/e2e/test_register.py index 5c3cc46bcdb..b2faa91b065 100644 --- a/tests/e2e/test_register.py +++ b/tests/e2e/test_register.py @@ -63,7 +63,6 @@ def basic_featuretable(): message_format=ProtoFormat(class_path="class.path"), topic="test_topic", event_timestamp_column="datetime_col", - created_timestamp_column="timestamp", ) return FeatureTable( name="basic_featuretable",