example to query kafka with SQL and visualize with marimo notebook

Author: jovezhong

examples/README.md

@@ -22,6 +22,8 @@ This folder lists some examples to run Timeplus Proton for various use cases. Fo
 
 - [jdbc](jdbc): demonstrates how to connect to Proton via JDBC using DBeaver or Metabase.
 
+- [marimo](marimo): demonstrates how to query Kafka data with SQL and visualize the results using Marimo.
+
 - [nginx-access-logs-streaming](nginx-access-logs-streaming) analyzing Nginx access logs using Timeplus Proton
 
 - [nginx-grafana](nginx-grafana) visualizing Nginx access logs using Timeplus and Grafana

examples/marimo/README.md

@@ -0,0 +1,10 @@
+# Example to Query Kafka Data with SQL and Visualize Results Using Marimo
+👋 This is a live notebook, built with [Timeplus](https://github.com/timeplus-io/proton) and [marimo](https://marimo.io), showing streaming data from GitHub via a public facing Kafka broker from [Aiven](https://aiven.io).
+
+Simply run the following commands:
+```bash
+curl https://astral.sh/uv/install.sh | sh
+curl https://install.timeplus.com/oss | sh
+./proton server&
+uvx marimo run --sandbox github.py
+```

examples/marimo/github.py

@@ -0,0 +1,232 @@
+# /// script
+# requires-python = ">=3.13"
+# dependencies = [
+#     "altair==5.5.0",
+#     "marimo",
+#     "polars[pyarrow]==1.26.0",
+#     "sqlalchemy==2.0.40",
+#     "sqlglot==26.12.1",
+#     "timeplus-connect==0.8.16",
+# ]
+# ///
+
+import marimo
+
+__generated_with = "0.12.5"
+app = marimo.App(width="medium")
+
+
+@app.cell(hide_code=True)
+def _(sqlalchemy):
+    engine = sqlalchemy.create_engine("timeplus://default:@localhost:8123")
+    return (engine,)
+
+
+@app.cell(hide_code=True)
+def _():
+    kafka_broker="kafka-public-read-timeplus.a.aivencloud.com:28864"
+    kafka_pwd="AVNS_MUaDRshCpeePa93AQy_"
+    return kafka_broker, kafka_pwd
+
+
+@app.cell(hide_code=True)
+def _(engine, kafka_broker, kafka_pwd, mo):
+    _df = mo.sql(
+        f"""
+        CREATE EXTERNAL STREAM IF NOT EXISTS github_events(actor string,
+                  created_at string,
+                  id string,
+                  payload string,
+                  repo string,
+                  type string
+                 )
+        SETTINGS type='kafka',
+                 brokers='{kafka_broker}',
+                 topic='github_events',
+                 security_protocol='SASL_SSL',
+                 sasl_mechanism='SCRAM-SHA-256',
+                 username='readonly',
+                 password='{kafka_pwd}',
+                 skip_ssl_cert_check=true,
+                 data_format='JSONEachRow',
+                 one_message_per_row=true
+        """,
+        output=False,
+        engine=engine
+    )
+    return
+
+
+@app.cell(hide_code=True)
+def _(kafka_broker, mo):
+    mo.md(
+        f"""
+        # Live GitHub Events
+        👋 This is a live notebook, built with [Timeplus](https://github.com/timeplus-io/proton) and [marimo](https://marimo.io), showing streaming data from GitHub via a public facing Kafka broker from [Aiven](https://aiven.io):
+        `{kafka_broker}`
+
+        Simply run the following commands:
+        ```bash
+        curl https://astral.sh/uv/install.sh | sh
+        curl https://install.timeplus.com/oss | sh
+        ./proton server&
+        uvx marimo run --sandbox github.py
+        ```
+        """
+    )
+    return
+
+
+@app.cell(hide_code=True)
+def _(mo):
+    range=mo.ui.slider(start=1, stop=10, step=1,label="View data for X minutes:", show_value=True)
+    return (range,)
+
+
+@app.cell
+def _(mo):
+    cntRefresh = mo.ui.refresh(options=["2s"],default_interval="2s")
+    return (cntRefresh,)
+
+
+@app.cell
+def _(df_cnt, get_count, mo, set_count):
+    _new_count=df_cnt["cnt"][0]
+    _diff=_new_count-get_count()
+    set_count(_new_count)
+    mo.stat(_new_count,label="In the Kafka topic",caption=f"{_diff} events", direction="increase" if _diff >= 0 else "decrease")
+    return
+
+
+@app.cell
+def _(df_last, mo):
+    from datetime import datetime, timezone
+    event_time=datetime.strptime(df_last["created_at"][0],"%Y-%m-%dT%H:%M:%S").replace(tzinfo=timezone.utc)
+    current_time = datetime.now(timezone.utc)
+    mo.stat(f"{datetime.strftime(event_time,"%H:%M:%S %m/%d")} (UTC)",label="Last Event",caption=f"{int((current_time - event_time).total_seconds()/60)} mins ago")
+    return current_time, datetime, event_time, timezone
+
+
+@app.cell
+def _(mo):
+    refresh = mo.ui.refresh(label="Refresh",options=["5s", "10s", "30s"])
+    return (refresh,)
+
+
+@app.cell
+def _(chart_repos, chart_types, mo, range, refresh):
+    mo.vstack([mo.hstack([range,refresh]),mo.hstack([chart_types,chart_repos],widths=[0,1])])
+    return
+
+
+@app.cell(hide_code=True)
+def _(chart_types):
+    _type=' '
+    if chart_types.selections.get("select_point"):
+        _array=chart_types.selections["select_point"].get("type",None)
+        if _array:
+            _type=f"WHERE type='{_array[0]}'"
+    typeWhere=_type
+    return (typeWhere,)
+
+
+@app.cell
+def _(mo):
+    mo.md(r"""## Sample Events""")
+    return
+
+
+@app.cell
+def _(engine, mo, refresh):
+    df_last = mo.sql(
+        f"""
+        -- {refresh.value}
+        SELECT * FROM github_events order by _tp_time desc limit 3 settings seek_to='-10s'
+        """,
+        engine=engine
+    )
+    return (df_last,)
+
+
+@app.cell(hide_code=True)
+def _():
+    import marimo as mo
+    import sqlalchemy
+    import altair as alt
+    return alt, mo, sqlalchemy
+
+
+@app.cell
+def _(alt, df_type, mo):
+    chart_types = mo.ui.altair_chart(
+        alt.Chart(df_type, height=150, width=150)
+        .mark_arc()
+        .encode(theta="cnt", color="type"),
+        legend_selection=False
+    )
+    return (chart_types,)
+
+
+@app.cell
+def _(alt, df_hotrepo, mo):
+    chart_repos = mo.ui.altair_chart(
+        alt.Chart(df_hotrepo, height=200)
+        .mark_bar()
+        .encode(x='cnt',
+                y=alt.Y('repo',sort=alt.EncodingSortField(field='cnt',order='descending')),)
+    )
+    return (chart_repos,)
+
+
+@app.cell
+def _(engine, mo, range, refresh, typeWhere):
+    df_hotrepo = mo.sql(
+        f"""
+        -- {refresh.value}
+        with cte as(SELECT top_k(repo,10,true) as a FROM github_events {typeWhere} limit 1 SETTINGS seek_to='-{range.value}m')
+        select a.1 as repo,a.2 as cnt from cte array join a
+        """,
+        engine=engine
+    )
+    return (df_hotrepo,)
+
+
+@app.cell(hide_code=True)
+def _(engine, mo, range, refresh):
+    df_type = mo.sql(
+        f"""
+        -- {refresh.value}
+        with cte as(SELECT top_k(type,10,true) as a FROM github_events limit 1 SETTINGS seek_to='-{range.value}m')
+        select a.1 as type,a.2 as cnt from cte array join a
+        """,
+        engine=engine
+    )
+    return (df_type,)
+
+
+@app.cell
+def _(cntRefresh):
+    cntRefresh.style({"display": None})
+    return
+
+
+@app.cell(hide_code=True)
+def cell_cnt(cntRefresh, engine, mo):
+    df_cnt = mo.sql(
+        f"""
+        -- {cntRefresh.value}
+        SELECT count() as cnt FROM github_events
+        """,
+        engine=engine
+    )
+    return (df_cnt,)
+
+
+@app.cell
+def _(mo):
+    get_count, set_count = mo.state(0)
+    return get_count, set_count
+
+
+if __name__ == "__main__":
+    app.run()