Rework sources and sinks - WIP

clickstream-ai-recommendation/README.md

@@ -36,7 +36,7 @@ mutation {
   }
 }
 ```
-You can find more content samples to add in the [sample_content.jsonl](content-api/sample_content.jsonl) file.
+You can find more content samples to add in the [sample_content.jsonl](connectors/testdata/sample_content.jsonl) file.
 
 Once you have added multiple pieces of content, we can register clicks with this mutation:
 ```graphql
@@ -85,10 +85,10 @@ docker run -it -p 8081:8081 -p 8888:8888 -p 9092:9092 --rm -v $PWD:/build -e OPE
 1. Once everything is started, open another terminal window to add data to Kafka using the
    load_data.py script in the `yourdata-files` directory. This requires **kafka-python-ng** installed
    via `pip3 install kafka-python-ng`.
-1. Load the content data: `python3 load_data.py content.json.gz localhost:9092 content --msg 50`.
+2. Load the content data: `python3 load_data.py content.json.gz localhost:9092 content --msg 50`.
    Wait until it finishes, which takes about two minutes. Check the Flink Dashboard running
    at http://localhost:8081/ to see the progress. Wait until the task turns blue again.
-1. Load the clickstream
+3. Load the clickstream
    data: `python3 load_data.py clickstream.json.gz localhost:9092 clickstream --msg 100`. This loads
    100 clicks per second. Wait a few seconds for some data to load. Let this run in the background
    until it finishes (about 4 minutes).

clickstream-ai-recommendation/connectors/sources-api.sqrl

@@ -0,0 +1,12 @@
+CREATE TABLE _Clickstream (
+                             url STRING NOT NULL,
+                             userid STRING NOT NULL,
+                             event_time TIMESTAMP_LTZ(3) NOT NULL METADATA FROM 'timestamp'
+);
+
+CREATE TABLE _Content (
+                         url STRING NOT NULL,
+                         title STRING NOT NULL,
+                         text STRING NOT NULL,
+                         update_time TIMESTAMP_LTZ(3) NOT NULL METADATA FROM 'timestamp'
+)

clickstream-ai-recommendation/connectors/sources-kafka.sqrl

@@ -0,0 +1,30 @@
+CREATE TABLE _Clickstream (
+                             url STRING NOT NULL,
+                             userid STRING NOT NULL,
+                             event_time TIMESTAMP_LTZ(3) NOT NULL METADATA FROM 'timestamp',
+                             WATERMARK FOR event_time AS event_time - INTERVAL '0.001' SECOND
+) WITH (
+      'connector' = 'kafka',
+      'topic' = 'clickstream',
+      'properties.bootstrap.servers' = '${KAFKA_BOOTSTRAP_SERVERS}',
+      'properties.group.id' = 'clickstream-group',
+      'scan.startup.mode' = 'group-offsets',
+      'properties.auto.offset.reset' = 'earliest',
+      'format' = 'flexible-json'
+      );
+
+CREATE TABLE _Content (
+                         url STRING NOT NULL,
+                         title STRING NOT NULL,
+                         text STRING NOT NULL,
+                         update_time TIMESTAMP_LTZ(3) NOT NULL METADATA FROM 'timestamp',
+                         WATERMARK FOR update_time AS update_time - INTERVAL '0.001' SECOND
+) WITH (
+      'connector' = 'kafka',
+      'topic' = 'content',
+      'properties.bootstrap.servers' = '${KAFKA_BOOTSTRAP_SERVERS}',
+      'properties.group.id' = 'clickstream-group',
+      'scan.startup.mode' = 'group-offsets',
+      'properties.auto.offset.reset' = 'earliest',
+      'format' = 'flexible-json'
+      );

clickstream-ai-recommendation/recommendation.sqrl

@@ -1,13 +1,9 @@
-/*+no_query */
-IMPORT clickstream-{{variant}}.clickstream;
-
-/*+no_query */
-IMPORT content-{{variant}}.content;
+IMPORT connectors.sources-{{variant}} AS sources;
 
 /* Find next page visits within 10 minutes */
 _CoVisits := SELECT b.url AS beforeURL, a.url AS afterURL,
                    a.event_time AS `timestamp`
-            FROM Clickstream b INNER JOIN Clickstream a ON b.userid=a.userid
+            FROM sources._Clickstream b INNER JOIN sources._Clickstream a ON b.userid=a.userid
                    AND b.event_time < a.event_time AND
                        b.event_time >= a.event_time - INTERVAL 10 MINUTE;
 /* Recommend pages that are visited shortly after */
@@ -19,14 +15,14 @@ Recommendation := SELECT beforeURL AS url, afterURL AS recommendation,
 
 IMPORT stdlib.openai.vector_embed;
 
-_ContentEmbedding := SELECT *, vector_embed(text, 'text-embedding-3-small') AS embedding FROM Content;
+_ContentEmbedding := SELECT *, vector_embed(text, 'text-embedding-3-small') AS embedding FROM sources._Content;
 
 EXPORT _ContentEmbedding TO logger.content;
 
 /*+vector_dim(embedding, 1536), index(VECTOR_COSINE, embedding) */
 _DistinctContent := DISTINCT _ContentEmbedding ON url ORDER BY update_time DESC;
 
-_UserInterest := SELECT userid, CENTER(embedding) AS interest FROM Clickstream click
+_UserInterest := SELECT userid, CENTER(embedding) AS interest FROM sources._Clickstream click
                 JOIN _DistinctContent FOR SYSTEM_TIME AS OF click.event_time content ON content.url = click.url
                 GROUP BY userid;
 

finance-credit-card-chatbot/credit-card-analytics/connectors/sources-kafka.sqrl

@@ -0,0 +1,69 @@
+CREATE TABLE CardAssignment (
+    customerId BIGINT NOT NULL,
+    cardNo STRING NOT NULL,
+    cardType STRING NOT NULL,
+    `timestamp` TIMESTAMP_LTZ(3) NOT NULL METADATA FROM 'timestamp',
+    WATERMARK FOR `timestamp` AS `timestamp` - INTERVAL '1' SECOND
+) WITH (
+      'connector' = 'kafka',
+      'properties.bootstrap.servers' = '${KAFKA_BOOTSTRAP_SERVERS}',
+      'properties.group.id' = 'mygroupid',
+      'scan.startup.mode' = 'group-offsets',
+      'properties.auto.offset.reset' = 'earliest',
+      'value.format' = 'flexible-json',
+      'topic' = 'cardassignment'
+      );
+
+CREATE TABLE Merchant (
+    merchantId BIGINT NOT NULL,
+    name STRING NOT NULL,
+    category STRING NOT NULL,
+    updatedTime TIMESTAMP_LTZ(3) NOT NULL  METADATA FROM 'timestamp',
+    WATERMARK FOR `updatedTime` AS `updatedTime` - INTERVAL '1' SECOND
+) WITH (
+      'connector' = 'kafka',
+      'properties.bootstrap.servers' = '${KAFKA_BOOTSTRAP_SERVERS}',
+      'properties.group.id' = 'mygroupid',
+      'scan.startup.mode' = 'group-offsets',
+      'properties.auto.offset.reset' = 'earliest',
+      'value.format' = 'flexible-json',
+      'topic' = 'merchant'
+      );
+
+
+CREATE TABLE MerchantReward (
+    merchantId BIGINT NOT NULL,
+    rewardsByCard ARRAY<ROW<
+        cardType STRING,
+        rewardPercentage BIGINT,
+        startTimestamp BIGINT,
+        expirationTimestamp BIGINT
+        >> NOT NULL,
+    updatedTime TIMESTAMP_LTZ(3) NOT NULL METADATA FROM 'timestamp',
+    WATERMARK FOR `updatedTime` AS `updatedTime` - INTERVAL '1' SECOND
+) WITH (
+      'connector' = 'kafka',
+      'properties.bootstrap.servers' = '${KAFKA_BOOTSTRAP_SERVERS}',
+      'properties.group.id' = 'mygroupid',
+      'scan.startup.mode' = 'group-offsets',
+      'properties.auto.offset.reset' = 'earliest',
+      'value.format' = 'flexible-json',
+      'topic' = 'merchantreward'
+      );
+
+CREATE TABLE Transaction (
+    transactionId BIGINT NOT NULL,
+    cardNo STRING NOT NULL,
+    `time` TIMESTAMP_LTZ(3) NOT NULL METADATA FROM 'timestamp',
+    amount DOUBLE NOT NULL,
+    merchantId BIGINT NOT NULL,
+    WATERMARK FOR `time` AS `time` - INTERVAL '1' SECOND
+) WITH (
+      'connector' = 'kafka',
+      'properties.bootstrap.servers' = '${KAFKA_BOOTSTRAP_SERVERS}',
+      'properties.group.id' = 'mygroupid',
+      'scan.startup.mode' = 'group-offsets',
+      'properties.auto.offset.reset' = 'earliest',
+      'value.format' = 'flexible-json',
+      'topic' = 'transaction'
+      );

finance-credit-card-chatbot/credit-card-analytics/connectors/sources-testdata.sqrl

@@ -0,0 +1,55 @@
+CREATE TABLE CardAssignment (
+    customerId BIGINT NOT NULL,
+    cardNo STRING NOT NULL,
+    cardType STRING NOT NULL,
+    `timestamp` TIMESTAMP_LTZ(3) NOT NULL,
+    PRIMARY KEY (`customerId`, `cardNo`, `timestamp`) NOT ENFORCED,
+    WATERMARK FOR `timestamp` AS `timestamp` - INTERVAL '1' SECOND
+) WITH (
+      'format' = 'flexible-json',
+      'path' = '${DATA_PATH}/card_assignment.jsonl',
+      'source.monitor-interval' = '10 min',
+      'connector' = 'filesystem'
+      );
+
+CREATE TABLE Merchant (
+    merchantId BIGINT NOT NULL,
+    name STRING NOT NULL,
+    category STRING NOT NULL,
+    updatedTime TIMESTAMP_LTZ(3) NOT NULL,
+    PRIMARY KEY (`merchantId`, `updatedTime`) NOT ENFORCED,
+    WATERMARK FOR `updatedTime` AS `updatedTime` - INTERVAL '1' SECOND
+) WITH (
+      'format' = 'flexible-json',
+      'path' = '${DATA_PATH}/merchant.jsonl',
+      'source.monitor-interval' = '10 min',
+      'connector' = 'filesystem'
+      );
+
+CREATE TABLE MerchantReward (
+    merchantId BIGINT NOT NULL,
+    rewardsByCard ARRAY<ROW<cardType STRING, rewardPercentage BIGINT, startTimestamp BIGINT, expirationTimestamp BIGINT>> NOT NULL,
+    updatedTime TIMESTAMP_LTZ(3) NOT NULL,
+    PRIMARY KEY (`merchantId`, `updatedTime`) NOT ENFORCED,
+    WATERMARK FOR `updatedTime` AS `updatedTime` - INTERVAL '1' SECOND
+) WITH (
+      'format' = 'flexible-json',
+      'path' = '${DATA_PATH}/merchant_reward.jsonl',
+      'source.monitor-interval' = '10 min',
+      'connector' = 'filesystem'
+      );
+
+CREATE TABLE Transaction (
+    transactionId BIGINT NOT NULL,
+    cardNo STRING NOT NULL,
+    `time` TIMESTAMP_LTZ(3) NOT NULL,
+    amount DOUBLE NOT NULL,
+    merchantId BIGINT NOT NULL,
+    PRIMARY KEY (`transactionId`, `time`) NOT ENFORCED,
+    WATERMARK FOR `time` AS `time` - INTERVAL '1' SECOND
+) WITH (
+      'format' = 'flexible-json',
+      'path' = '${DATA_PATH}/transaction.jsonl',
+      'source.monitor-interval' = '10 min',
+      'connector' = 'filesystem'
+      );