YI FU fuyi

Code to write to Feature Store

@beam.typehints.with_input_types(Tuple[User, List[str]])
class WriteFeatureRPC(beam.DoFn):
    def __init__(
        self,
        api_endpoint,
        project_id,

Create custom window strategy to fulfil sequencer model feature requirement

Background

To train our neural sequencer model properly, we need to get the ordered sequence of user interaction with articles in a certain period of time. To formulate the requirements, the input dataset for the model training can be described as:

The last X articles the user has interacted with in the last Y days in chronological order.

X: variable with integer value between [0, 20] inclusive

tail -400 /srv/hops/onlinefs/logs/onlinefs.log

Enable succeeded: 
[stdout]
-b083-4198-9ae5-d966137a8076 sending LeaveGroup request to coordinator 10.0.0.9:9091 (id: 2147483646 rack: null)
2021-08-23 14:30:52 INFO  AbstractCoordinator:879 - [Consumer clientId=consumer-3, groupId=0] Member consumer-3-5d5705ab-e091-4f8a-a331-974f410e4273 sending LeaveGroup request to coordinator 10.0.0.9:9091 (id: 2147483646 rack: null)

	# use C-a, since it's on the home row and easier to hit than C-b
	set-option -g prefix C-a
	unbind-key C-a
	bind-key C-a send-prefix
	set -g base-index 1

	# Easy config reload
	bind-key R source-file ~/.tmux.conf \; display-message "tmux.conf reloaded."

	# vi is good

	FROM public.ecr.aws/lambda/python:3.8
	RUN yum -y install java-1.8.0-openjdk wget curl
	COPY requirements.txt requirements.txt
	RUN pip install -r requirements.txt
	ENV JAVA_HOME="/usr/lib/jvm/java-1.8.0-openjdk-1.8.0.302.b08-0.amzn2.0.1.x86_64/jre"
	ENV PATH=${PATH}:${JAVA_HOME}/bin
	ENV SPARK_HOME="/var/lang/lib/python3.8/site-packages/pyspark"
	ENV PATH=$PATH:$SPARK_HOME/bin
	ENV PYTHONPATH=$SPARK_HOME/python:$SPARK_HOME/python/lib/py4j-0.10.9-src.zip:$PYTHONPATH
	ENV PATH=$SPARK_HOME/python:$PATH

	{
	"parameters":[
	{
	"name":"touchpoint",
	"stringValue":"android"
	},
	{
	"name":"market",
	"stringValue":"gb"
	},

	{
	"parameters": [
	{
	"name": "touchpoint",
	"value_string": "DESKTOP"
	},
	{
	"name": "market",
	"value_string": "SE"
	},

	---------------------------------------------------------------------------
	Py4JJavaError Traceback (most recent call last)
	<command-2690650308556142> in <module>
	5 primary_key=["customer_id"])
	6
	----> 7 add_to_cart_last_7days_fg.save(empty_add_to_cart_last_7days)

	/databricks/python/lib/python3.8/site-packages/hsfs/feature_group.py in save(self, features, write_options)
	642
	643 # fg_job is used only if the hive engine is used

	@beam.typehints.with_input_types(Tuple[User, List[EventAction]])
	class WriteFeatureRPC(beam.DoFn):
	def __init__(
	self,
	api_endpoint,
	project_id,
	location,
	featurestore_id,
	entity_type_id,
	feature_name,

	INPUT = [
	1,
	2,
	3,
	4,
	5,
	11111,
	9,
	8,
	7,