Join and Group Bys for POC

api/py/test/sample/group_bys/risk/merchant_data.py

@@ -0,0 +1,29 @@
+from ai.chronon.api.ttypes import Source, EntitySource
+from ai.chronon.query import Query, select
+from ai.chronon.group_by import (
+    GroupBy,
+    Aggregation,
+    Operation,
+    Window,
+    TimeUnit
+)
+
+"""
+This GroupBy aggregates metrics about a user's previous purchases in various windows.
+"""
+
+# This source is raw purchase events. Every time a user makes a purchase, it will be one entry in this source.
+source_merchants = Source(
+    entities=EntitySource(
+        snapshotTable="data.merchants", # This points to the log table in the warehouse with historical purchase events, updated in batch daily
+        query=Query(
+            selects=select("merchant_id","account_age", "zipcode", "is_big_merchant", "country", "account_type", "preferred_language"), # Select the fields we care about
+        )
+    )
+)
+
+merchant_group_by = GroupBy(
+    sources=[source_merchants],
+    keys=["merchant_id"],
+    aggregations=None
+)

api/py/test/sample/group_bys/risk/transaction_events.py

@@ -0,0 +1,53 @@
+from ai.chronon.api.ttypes import Source, EventSource
+from ai.chronon.query import Query, select
+from ai.chronon.group_by import (
+    GroupBy,
+    Aggregation,
+    Operation,
+    Window,
+    TimeUnit
+)
+
+"""
+This GroupBy aggregates metrics about a user's previous purchases in various windows.
+"""
+
+def create_transaction_source(key_field):
+    return Source(
+        events=EventSource(
+            table="data.txn_events",  # Points to the historical purchase events table
+            topic=None,
+            query=Query(
+                selects=select(key_field, "transaction_amount", "transaction_type"),
+                time_column="transaction_time"
+            )
+        )
+    )
+
+def create_txn_group_by(source, key):
+    return GroupBy(
+        sources=[source],
+        keys=[key],
+        online=True,
+        aggregations=[
+            Aggregation(
+                input_column="transaction_amount",
+                operation=Operation.COUNT,
+                windows=window_sizes
+            ),
+            Aggregation(
+                input_column="transaction_amount",
+                operation=Operation.SUM,
+                windows=[Window(length=1, timeUnit=TimeUnit.HOURS)]
+            )
+        ]
+    )
+
+
+window_sizes = [Window(length=1, timeUnit=TimeUnit.HOURS), Window(length=1, timeUnit=TimeUnit.DAYS), Window(length=30, timeUnit=TimeUnit.DAYS), Window(length=365, timeUnit=TimeUnit.DAYS)]
+
+source_user_transactions = create_transaction_source("user_id")
+txn_group_by_user = create_txn_group_by(source_user_transactions, "user_id")
+
+source_merchant_transactions = create_transaction_source("merchant_id")
+txn_group_by_merchant = create_txn_group_by(source_merchant_transactions, "merchant_id")

api/py/test/sample/group_bys/risk/user_data.py

@@ -0,0 +1,29 @@
+from ai.chronon.api.ttypes import Source, EntitySource
+from ai.chronon.query import Query, select
+from ai.chronon.group_by import (
+    GroupBy,
+    Aggregation,
+    Operation,
+    Window,
+    TimeUnit
+)
+
+"""
+This GroupBy aggregates metrics about a user's previous purchases in various windows.
+"""
+
+# This source is raw purchase events. Every time a user makes a purchase, it will be one entry in this source.
+source_users = Source(
+    entities=EntitySource(
+        snapshotTable="data.users", # This points to the log table in the warehouse with historical purchase events, updated in batch daily
+        query=Query(
+            selects=select("user_id","account_age", "account_balance", "credit_score", "number_of_devices", "country", "account_type", "preferred_language"), # Select the fields we care about
+        ) # The event time
+    )
+)
+
+user_group_by = GroupBy(
+    sources=[source_users],
+    keys=["user_id"],
+    aggregations=None
+)

api/py/test/sample/joins/risk/user_transactions.py

@@ -0,0 +1,21 @@
+from ai.chronon.api.ttypes import Source, EventSource
+from ai.chronon.join import Join, JoinPart
+from ai.chronon.query import Query, select
+from group_bys.risk.transaction_events import txn_group_by_user, txn_group_by_merchant
+from group_bys.risk.user_data import user_group_by
+from group_bys.risk.merchant_data import merchant_group_by
+
+source_users = Source(
+    events=EventSource(
+        table="data.users",
+        query=Query(
+            selects=select("user_id"),
+            time_column="ts"
+        )
+    )
+)
+
+txn_join = Join(
+    left=source_users,
+    right_parts=[JoinPart(group_by=txn_group_by_user, prefix="user"), JoinPart(group_by=txn_group_by_merchant, prefix="merchant"), JoinPart(group_by=user_group_by, prefix="user"), JoinPart(group_by=merchant_group_by, prefix="merchant")]
+)

api/py/test/sample/production/group_bys/risk/transaction_events.txn_group_by_merchant

@@ -0,0 +1,71 @@
+{
+  "metaData": {
+    "name": "risk.transaction_events.txn_group_by_merchant",
+    "online": 1,
+    "customJson": "{\"lag\": 0, \"groupby_tags\": null, \"column_tags\": {}}",
+    "dependencies": [
+      "{\"name\": \"wait_for_data.txn_events_ds\", \"spec\": \"data.txn_events/ds={{ ds }}\", \"start\": null, \"end\": null}"
+    ],
+    "tableProperties": {
+      "source": "chronon"
+    },
+    "outputNamespace": "default",
+    "team": "risk",
+    "offlineSchedule": "@daily"
+  },
+  "sources": [
+    {
+      "events": {
+        "table": "data.txn_events",
+        "query": {
+          "selects": {
+            "merchant_id": "merchant_id",
+            "transaction_amount": "transaction_amount",
+            "transaction_type": "transaction_type"
+          },
+          "timeColumn": "transaction_time",
+          "setups": []
+        }
+      }
+    }
+  ],
+  "keyColumns": [
+    "merchant_id"
+  ],
+  "aggregations": [
+    {
+      "inputColumn": "transaction_amount",
+      "operation": 6,
+      "argMap": {},
+      "windows": [
+        {
+          "length": 1,
+          "timeUnit": 0
+        },
+        {
+          "length": 1,
+          "timeUnit": 1
+        },
+        {
+          "length": 30,
+          "timeUnit": 1
+        },
+        {
+          "length": 365,
+          "timeUnit": 1
+        }
+      ]
+    },
+    {
+      "inputColumn": "transaction_amount",
+      "operation": 7,
+      "argMap": {},
+      "windows": [
+        {
+          "length": 1,
+          "timeUnit": 0
+        }
+      ]
+    }
+  ]
+}

api/py/test/sample/production/group_bys/risk/transaction_events.txn_group_by_user

@@ -0,0 +1,71 @@
+{
+  "metaData": {
+    "name": "risk.transaction_events.txn_group_by_user",
+    "online": 1,
+    "customJson": "{\"lag\": 0, \"groupby_tags\": null, \"column_tags\": {}}",
+    "dependencies": [
+      "{\"name\": \"wait_for_data.txn_events_ds\", \"spec\": \"data.txn_events/ds={{ ds }}\", \"start\": null, \"end\": null}"
+    ],
+    "tableProperties": {
+      "source": "chronon"
+    },
+    "outputNamespace": "default",
+    "team": "risk",
+    "offlineSchedule": "@daily"
+  },
+  "sources": [
+    {
+      "events": {
+        "table": "data.txn_events",
+        "query": {
+          "selects": {
+            "user_id": "user_id",
+            "transaction_amount": "transaction_amount",
+            "transaction_type": "transaction_type"
+          },
+          "timeColumn": "transaction_time",
+          "setups": []
+        }
+      }
+    }
+  ],
+  "keyColumns": [
+    "user_id"
+  ],
+  "aggregations": [
+    {
+      "inputColumn": "transaction_amount",
+      "operation": 6,
+      "argMap": {},
+      "windows": [
+        {
+          "length": 1,
+          "timeUnit": 0
+        },
+        {
+          "length": 1,
+          "timeUnit": 1
+        },
+        {
+          "length": 30,
+          "timeUnit": 1
+        },
+        {
+          "length": 365,
+          "timeUnit": 1
+        }
+      ]
+    },
+    {
+      "inputColumn": "transaction_amount",
+      "operation": 7,
+      "argMap": {},
+      "windows": [
+        {
+          "length": 1,
+          "timeUnit": 0
+        }
+      ]
+    }
+  ]
+}