update

Author: Andy Hattemer

README.md

@@ -4,7 +4,7 @@
 
 This is a quick proof-of-concept of how [Cube](https://cube.dev/) integrates with [Materialize]()
 
-- **Materialize** is a streaming database that 
+- **Materialize** is a streaming database that
 - **Cube** is a "headless BI" service that connects to databases or data warehouses and handles Data Modeling, Access Control, Caching and APIs
 
 ## Starting Up
@@ -24,24 +24,16 @@ You'll need to have [docker and docker-compose installed](https://materialize.co
    docker-compose up -d
    ```
 
-   **This may take several minutes to complete the first time you run it.** If all goes well, you'll have everything running in their own containers.
+   **This may take several minutes to complete the first time you run it.** If all goes well, you'll have everything running in their own containers, with Debezium configured to ship changes from MySQL into Redpanda.
 
-3. Create the initial schema of sources and materialized views in Materialize
+3. Confirm that everything is running as expected:
 
    ```shell session
-   psql -h localhost -p 6875 -U materialize -f materialize/create.sql
+   docker-compose ps
    ```
 
-4. Open http://localhost:4000/ in your browser and connect Cube.js to materialize:
-   ![image](https://user-images.githubusercontent.com/11527560/168123759-e63489d8-12d9-47c0-bd8b-493632370075.png)
-   For the credentials, use:
-   | Field    | Value          |
-   |----------|----------------|
-   | host     | `materialized` |
-   | port     | `6875`         |
-   | database | `materialize`  |
-   | username | `materialize`  |
-   | password | _leave blank_  |
-   
-
-5. Three example models have been pre-created in the `cube/schema` directory, which is mounted as a volume so the Cube docker image recognizes and loads them.
+4. Initialize the Materialize Schema
+
+   ```shell session
+   psql -h localhost -p 6875 -U materialize -f materialize/create.sql
+   ```

compose.yaml

@@ -5,6 +5,12 @@ services:
       - 4000:4000
     environment:
       - CUBEJS_DEV_MODE=true
+      - CUBEJS_DB_TYPE=materialize
+      - CUBEJS_DB_HOST=materialized
+      - CUBEJS_DB_PORT=6875
+      - CUBEJS_DB_NAME=materialize
+      - CUBEJS_DB_USER=materialize
+      - CUBEJS_API_SECRET=SECRET
     volumes:
       - ${PWD}/cube:/cube/conf
   materialized:

cube/schema/.ItemSummary.js.swp

@@ -0,0 +1,50 @@
+cube(`Vendors`, {
+  refreshKey: {
+    every: `1 second`
+  },
+
+  sql: `SELECT * FROM public.agg_vendors_minute`,
+  measures: {
+    count: {
+      type: `count`,
+      drillMembers: [CUBE.name]
+    },
+    totalRevenue: {
+      type: `sum`,
+      sql: `revenue`,
+      format: `currency`
+    },
+    totalOrders: {
+      type: `sum`,
+      sql: `orders`
+    },
+    totalPageviews: {
+      type: `sum`,
+      sql: `pageviews`
+    },
+    totalItemsSold: {
+      type: `sum`,
+      sql: `items_sold`
+    },
+    conversionRate: {
+      type: `number`,
+      sql: `SUM(orders)/SUM(pageviews)+1`,
+      format: `percent`
+    },
+  },
+  dimensions: {
+    id: {
+      sql: `vendor_id`,
+      type: `number`
+    },
+    name: {
+      sql: `vendor_name`,
+      type: `string`
+    },
+    receivedAtMinute: {
+      sql: `m`,
+      type: `time`
+    }
+  },
+  dataSource: `default`,
+});

cube/schema/ItemSummary.js

@@ -0,0 +1,10 @@
+FROM python:3.9.9-bullseye
+
+WORKDIR /workdir
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY . .
+
+ENTRYPOINT ["python", "generate_load.py"]

cube/schema/ProfileViewsEnriched.js

@@ -0,0 +1,150 @@
+import json
+import math
+import os
+import random
+import time
+
+import barnum
+from kafka import KafkaProducer
+from mysql.connector import Error, connect
+
+# CONFIG
+vendorSeedCount = 100
+userSeedCount = 10000
+itemSeedCount = 1000
+purchaseGenCount = 5000000
+purchaseGenEveryMS = 100
+pageviewMultiplier = 75  # Translates to 75x purchases, currently 750/sec or 65M/day
+itemInventoryMin = 1000
+itemInventoryMax = 5000
+itemPriceMin = 5
+itemPriceMax = 500
+mysqlHost = "mysql"
+mysqlPort = "3306"
+mysqlUser = "mysqluser"
+mysqlPass = "mysqlpw"
+kafkaHostPort = os.getenv("KAFKA_ADDR", "redpanda:9092")
+kafkaTopic = "pageviews"
+debeziumHostPort = "debezium:8083"
+channels = ["organic search", "paid search", "referral", "social", "display"]
+categories = ["widgets", "gadgets", "doodads", "clearance"]
+
+# INSERT TEMPLATES
+vendor_insert = "INSERT INTO shop.vendors (name) VALUES ( %s )"
+item_insert = "INSERT INTO shop.items (vendor_id, name, category, price, inventory) VALUES ( %s, %s, %s, %s, %s )"
+user_insert = "INSERT INTO shop.users (email, is_vip) VALUES ( %s, %s )"
+purchase_insert = "INSERT INTO shop.purchases (user_id, item_id, quantity, purchase_price) VALUES ( %s, %s, %s, %s )"
+
+# Initialize Kafka
+producer = KafkaProducer(
+    bootstrap_servers=[kafkaHostPort],
+    value_serializer=lambda x: json.dumps(x).encode("utf-8"),
+)
+
+def generatePageview(viewer_id, target_id, page_type):
+    return {
+        "user_id": viewer_id,
+        "url": f"/{page_type}/{target_id}",
+        "channel": random.choice(channels),
+        "received_at": int(time.time()),
+    }
+
+
+try:
+    with connect(
+        host=mysqlHost,
+        user=mysqlUser,
+        password=mysqlPass,
+    ) as connection:
+        with connection.cursor() as cursor:
+            print("Seeding data...")
+            print(" - Vendors")
+            cursor.executemany(
+                vendor_insert,
+                [
+                    (
+                        [barnum.create_company_name()]
+                    )
+                    for i in range(vendorSeedCount)
+                ],
+            )
+            print(" - Items")
+            cursor.executemany(
+                item_insert,
+                [
+                    (
+                        random.randint(1, vendorSeedCount),
+                        barnum.create_nouns(),
+                        random.choice(categories),
+                        random.randint(itemPriceMin * 100, itemPriceMax * 100) / 100,
+                        random.randint(itemInventoryMin, itemInventoryMax),
+                    )
+                    for i in range(itemSeedCount)
+                ],
+            )
+            cursor.executemany(
+                user_insert,
+                [
+                    (barnum.create_email(), (random.randint(0, 10) > 8))
+                    for i in range(userSeedCount)
+                ],
+            )
+            connection.commit()
+
+            print("Getting item ID and PRICEs...")
+            cursor.execute("SELECT id, price FROM shop.items")
+            item_prices = [(row[0], row[1]) for row in cursor]
+
+            print("Preparing to loop + seed kafka pageviews and purchases")
+            for i in range(purchaseGenCount):
+                # Get a user and item to purchase
+                purchase_item = random.choice(item_prices)
+                purchase_user = random.randint(0, userSeedCount - 1)
+                purchase_quantity = random.randint(1, 5)
+
+                # Write purchaser pageview
+                producer.send(
+                    kafkaTopic,
+                    key=str(purchase_user).encode("ascii"),
+                    value=generatePageview(purchase_user, purchase_item[0], "products"),
+                )
+
+                # Write random pageviews to products or profiles
+                pageviewOscillator = int(
+                    pageviewMultiplier + (math.sin(time.time() / 1000) * 50)
+                )
+                for i in range(pageviewOscillator):
+                    rand_user = random.randint(0, userSeedCount)
+                    rand_page_type = random.choice(["products", "profiles"])
+                    target_id_max_range = (
+                        itemSeedCount if rand_page_type == "products" else userSeedCount
+                    )
+                    producer.send(
+                        kafkaTopic,
+                        key=str(rand_user).encode("ascii"),
+                        value=generatePageview(
+                            rand_user,
+                            random.randint(0, target_id_max_range),
+                            rand_page_type,
+                        ),
+                    )
+
+                # Write purchase row
+                cursor.execute(
+                    purchase_insert,
+                    (
+                        purchase_user,
+                        purchase_item[0],
+                        purchase_quantity,
+                        purchase_item[1] * purchase_quantity,
+                    ),
+                )
+                connection.commit()
+
+                # Pause
+                time.sleep(purchaseGenEveryMS / 1000)
+
+    connection.close()
+
+except Error as e:
+    print(e)