Redis online store materialization fails with UnicodeDecodeError: 'utf-8' codec can't decode byte 0xb4 in position 1: invalid start byte

[seekerofsai]

Expected Behavior

Snowflake to redis materialization succeeds.

Current Behavior

Materializing 2 feature views from 2021-09-10 00:00:00+00:00 to 2021-09-11 00:00:00+00:00 into the redis online store.

credit_scores_features:
Snowflake: Processing Materialization ResultSet Batch #1
0%| | 0/548 [00:00<?, ?it/s]

UnicodeDecodeError Traceback (most recent call last)

File /opt/conda/lib/python3.8/site-packages/feast/usage.py:288, in log_exceptions_and_usage..decorator..wrapper(*args, **kwargs)
285 ctx.attributes.update(attrs)
287 try:
--> 288 return func(*args, **kwargs)
289 except Exception:
290 if ctx.exception:
291 # exception was already recorded

File /opt/conda/lib/python3.8/site-packages/feast/feature_store.py:1395, in FeatureStore.materialize(self, start_date, end_date, feature_views)
1392 start_date = utils.make_tzaware(start_date)
1393 end_date = utils.make_tzaware(end_date)
-> 1395 provider.materialize_single_feature_view(
1396 config=self.config,
1397 feature_view=feature_view,
1398 start_date=start_date,
1399 end_date=end_date,
1400 registry=self._registry,
1401 project=self.project,
1402 tqdm_builder=tqdm_builder,
1403 )
1405 self._registry.apply_materialization(
1406 feature_view,
1407 self.project,
1408 start_date,
1409 end_date,
1410 )

File /opt/conda/lib/python3.8/site-packages/feast/infra/passthrough_provider.py:254, in PassthroughProvider.materialize_single_feature_view(self, config, feature_view, start_date, end_date, registry, project, tqdm_builder)
252 e = jobs[0].error()
253 assert e
--> 254 raise e

File /opt/conda/lib/python3.8/site-packages/feast/infra/materialization/snowflake_engine.py:299, in SnowflakeMaterializationEngine._materialize_one(self, registry, feature_view, start_date, end_date, project, tqdm_builder)
292 self.materialize_to_snowflake_online_store(
293 self.repo_config,
294 fv_to_proto_sql,
295 feature_view,
296 project,
297 )
298 else:
--> 299 self.materialize_to_external_online_store(
300 self.repo_config,
301 fv_to_proto_sql,
302 feature_view,
303 tqdm_builder,
304 )
306 return SnowflakeMaterializationJob(
307 job_id=job_id, status=MaterializationJobStatus.SUCCEEDED
308 )
309 except BaseException as e:

File /opt/conda/lib/python3.8/site-packages/feast/infra/materialization/snowflake_engine.py:498, in SnowflakeMaterializationEngine.materialize_to_external_online_store(self, repo_config, materialization_sql, feature_view, tqdm_builder)
488 rows_to_write = list(
489 zip(
490 entity_keys,
(...)
494 )
495 )
497 with tqdm_builder(len(rows_to_write)) as pbar:
--> 498 self.online_store.online_write_batch(
499 repo_config,
500 feature_view,
501 rows_to_write,
502 lambda x: pbar.update(x),
503 )
504 return None

File /opt/conda/lib/python3.8/site-packages/feast/usage.py:299, in log_exceptions_and_usage..decorator..wrapper(*args, **kwargs)
296 ctx.traceback = _trace_to_log(traceback)
298 if traceback:
--> 299 raise exc.with_traceback(traceback)
301 raise exc
302 finally:

File /opt/conda/lib/python3.8/site-packages/feast/usage.py:288, in log_exceptions_and_usage..decorator..wrapper(*args, **kwargs)
285 ctx.attributes.update(attrs)
287 try:
--> 288 return func(*args, **kwargs)
289 except Exception:
290 if ctx.exception:
291 # exception was already recorded

File /opt/conda/lib/python3.8/site-packages/feast/infra/online_stores/redis.py:219, in RedisOnlineStore.online_write_batch(self, config, table, data, progress)
217 keys.append(redis_key_bin)
218 pipe.hmget(redis_key_bin, ts_key)
--> 219 prev_event_timestamps = pipe.execute()
220 # flattening the list of lists. hmget does the lookup assuming a list of keys in the key bin
221 prev_event_timestamps = [i[0] for i in prev_event_timestamps]

File /opt/conda/lib/python3.8/site-packages/redis/cluster.py:1898, in ClusterPipeline.execute(self, raise_on_error)
1896 stack = self.command_stack
1897 try:
-> 1898 return self.send_cluster_commands(stack, raise_on_error)
1899 finally:
1900 self.reset()

File /opt/conda/lib/python3.8/site-packages/redis/cluster.py:1957, in ClusterPipeline.send_cluster_commands(self, stack, raise_on_error, allow_redirections)
1955 for _ in range(0, self.cluster_error_retry_attempts):
1956 try:
-> 1957 return self._send_cluster_commands(
1958 stack,
1959 raise_on_error=raise_on_error,
1960 allow_redirections=allow_redirections,
1961 )
1962 except ClusterDownError:
1963 # Try again with the new cluster setup. All other errors
1964 # should be raised.
1965 pass

File /opt/conda/lib/python3.8/site-packages/redis/cluster.py:2025, in ClusterPipeline._send_cluster_commands(self, stack, raise_on_error, allow_redirections)
2022 n.write()
2024 for n in node_commands:
-> 2025 n.read()
2027 # release all of the redis connections we allocated earlier
2028 # back into the connection pool.
2029 # we used to do this step as part of a try/finally block,
(...)
2043 # every single request after to that connection will always get
2044 # a mismatched result.
2045 for n in nodes.values():

File /opt/conda/lib/python3.8/site-packages/redis/cluster.py:2297, in NodeCommands.read(self)
2295 if c.result is None:
2296 try:
-> 2297 c.result = self.parse_response(connection, c.args[0], **c.options)
2298 except (ConnectionError, TimeoutError) as e:
2299 for c in self.commands:

File /opt/conda/lib/python3.8/site-packages/redis/client.py:1234, in Redis.parse_response(self, connection, command_name, **options)
1232 response = connection.read_response(disable_decoding=True)
1233 else:
-> 1234 response = connection.read_response()
1235 except ResponseError:
1236 if EMPTY_RESPONSE in options:

File /opt/conda/lib/python3.8/site-packages/redis/connection.py:821, in Connection.read_response(self, disable_decoding)
818 hosterr = "connection"
820 try:
--> 821 response = self._parser.read_response(disable_decoding=disable_decoding)
822 except socket.timeout:
823 self.disconnect()

File /opt/conda/lib/python3.8/site-packages/redis/connection.py:476, in HiredisParser.read_response(self, disable_decoding)
474 response = self._reader.gets(False)
475 else:
--> 476 response = self._reader.gets()
477 # if an older version of hiredis is installed, we need to attempt
478 # to convert ResponseErrors to their appropriate types.
479 if not HIREDIS_SUPPORTS_CALLABLE_ERRORS:

UnicodeDecodeError: 'utf-8' codec can't decode byte 0xb4 in position 1: invalid start byte

Steps to reproduce

Specifications

• Version: Feast 0.30.2
• Platform: Python 3.8
• Subsystem:

Possible Solution

[sfc-gh-madkins]

@seekerofsai I dont have enough here to diagnose the issue ... my guess is there might be a specific data type causing the issue here. Are you open to connecting to help me pinpoint what my be causing the error here?

[sfc-gh-madkins]

@seekerofsai does this work with the snowflake online store?

[seekerofsai]

Thanks for looking in to. Yes, it works with snowflake as online store. i was using aws elaticache redis cluster as online store. with trial and error i found that this error goes away when i leave these options default in connection string skip_full_coverage_check,decode_responses

[sfc-gh-madkins]

@seekerofsai can we close this issue out?

[seekerofsai]

issue specific to aws elasticache for redis workaround on configuration change works