From 679f24204f7df7facf23a9b6d34553f217b465b8 Mon Sep 17 00:00:00 2001 From: Oleksii Moskalenko Date: Wed, 20 Jan 2021 12:06:17 +0800 Subject: [PATCH 1/4] keep same amount of partitions Signed-off-by: Oleksii Moskalenko --- .../feast/ingestion/stores/redis/RedisSinkRelation.scala | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/spark/ingestion/src/main/scala/feast/ingestion/stores/redis/RedisSinkRelation.scala b/spark/ingestion/src/main/scala/feast/ingestion/stores/redis/RedisSinkRelation.scala index 3f3414b1760..8d63e6b0354 100644 --- a/spark/ingestion/src/main/scala/feast/ingestion/stores/redis/RedisSinkRelation.scala +++ b/spark/ingestion/src/main/scala/feast/ingestion/stores/redis/RedisSinkRelation.scala @@ -70,8 +70,8 @@ class RedisSinkRelation(override val sqlContext: SQLContext, config: SparkRedisC override def insert(data: DataFrame, overwrite: Boolean): Unit = { // repartition for deduplication val dataToStore = - if (config.repartitionByEntity) - data.repartition(config.entityColumns.map(col): _*).localCheckpoint() + if (config.repartitionByEntity && data.rdd.getNumPartitions > 1) + data.repartition(data.rdd.getNumPartitions, config.entityColumns.map(col): _*).localCheckpoint() else data dataToStore.foreachPartition { partition: Iterator[Row] => From 49ad61a908abe591fde3021b6d207475162e1cef Mon Sep 17 00:00:00 2001 From: Oleksii Moskalenko Date: Wed, 20 Jan 2021 12:13:08 +0800 Subject: [PATCH 2/4] fix formatting Signed-off-by: Oleksii Moskalenko --- .../feast/ingestion/stores/redis/RedisSinkRelation.scala | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/spark/ingestion/src/main/scala/feast/ingestion/stores/redis/RedisSinkRelation.scala b/spark/ingestion/src/main/scala/feast/ingestion/stores/redis/RedisSinkRelation.scala index 8d63e6b0354..2f6c758b5eb 100644 --- a/spark/ingestion/src/main/scala/feast/ingestion/stores/redis/RedisSinkRelation.scala +++ b/spark/ingestion/src/main/scala/feast/ingestion/stores/redis/RedisSinkRelation.scala @@ -71,7 +71,9 @@ class RedisSinkRelation(override val sqlContext: SQLContext, config: SparkRedisC // repartition for deduplication val dataToStore = if (config.repartitionByEntity && data.rdd.getNumPartitions > 1) - data.repartition(data.rdd.getNumPartitions, config.entityColumns.map(col): _*).localCheckpoint() + data + .repartition(data.rdd.getNumPartitions, config.entityColumns.map(col): _*) + .localCheckpoint() else data dataToStore.foreachPartition { partition: Iterator[Row] => From 5ef35046505e8f4b01ce3463c47ef304ec9db5eb Mon Sep 17 00:00:00 2001 From: Oleksii Moskalenko Date: Thu, 21 Jan 2021 10:23:40 +0800 Subject: [PATCH 3/4] debug Signed-off-by: Oleksii Moskalenko --- infra/scripts/test-docker-compose.sh | 1 + 1 file changed, 1 insertion(+) diff --git a/infra/scripts/test-docker-compose.sh b/infra/scripts/test-docker-compose.sh index e2454f9fe85..ad414770aeb 100755 --- a/infra/scripts/test-docker-compose.sh +++ b/infra/scripts/test-docker-compose.sh @@ -14,6 +14,7 @@ clean_up () { # Shut down docker-compose images + docker logs feast_jobservice_1 docker-compose down exit $ARG From 4519cb32bce4df3d10e97de1e3155a21c4b53f9d Mon Sep 17 00:00:00 2001 From: Oleksii Moskalenko Date: Thu, 21 Jan 2021 11:21:22 +0800 Subject: [PATCH 4/4] standalone launcher: just check that job started Signed-off-by: Oleksii Moskalenko --- infra/scripts/test-docker-compose.sh | 1 - sdk/python/feast/pyspark/launchers/standalone/local.py | 5 +---- 2 files changed, 1 insertion(+), 5 deletions(-) diff --git a/infra/scripts/test-docker-compose.sh b/infra/scripts/test-docker-compose.sh index ad414770aeb..e2454f9fe85 100755 --- a/infra/scripts/test-docker-compose.sh +++ b/infra/scripts/test-docker-compose.sh @@ -14,7 +14,6 @@ clean_up () { # Shut down docker-compose images - docker logs feast_jobservice_1 docker-compose down exit $ARG diff --git a/sdk/python/feast/pyspark/launchers/standalone/local.py b/sdk/python/feast/pyspark/launchers/standalone/local.py index e67dd756300..645b3e7ff16 100644 --- a/sdk/python/feast/pyspark/launchers/standalone/local.py +++ b/sdk/python/feast/pyspark/launchers/standalone/local.py @@ -124,10 +124,7 @@ def check_if_started(self): if not app: return False - stages = requests.get( - f"http://localhost:{self._ui_port}/api/v1/applications/{app['id']}/stages" - ).json() - return bool(stages) + return True def get_start_time(self) -> datetime: return self._start_time