Add a simple notebook to test kafka streaming

Author: incrypto32

notebooks/kafka_streaming.py

@@ -0,0 +1,164 @@
+import marimo
+
+__generated_with = "0.17.0"
+app = marimo.App(width="medium")
+
+
+@app.cell
+def _():
+    import marimo as mo
+    from amp.client import Client
+    return Client, mo
+
+
+@app.cell(hide_code=True)
+def _(mo):
+    mo.md(
+        r"""
+    # Kafka Streaming Example
+
+    This notebook demonstrates continuous streaming from Flight SQL to Kafka with reorg detection.
+    """
+    )
+    return
+
+
+@app.cell(hide_code=True)
+def _(mo):
+    mo.md(r"""## Setup""")
+    return
+
+
+@app.cell
+def _(Client):
+    client = Client('grpc://34.122.177.97:80')
+    return (client,)
+
+
+@app.cell
+def _(client):
+    client.configure_connection(
+        'my_kafka',
+        'kafka',
+        {
+            'bootstrap_servers': 'localhost:9092',
+            'client_id': 'amp-streaming-client',
+            'key_field': 'block_num'
+        }
+    )
+    return
+
+
+@app.cell(hide_code=True)
+def _(mo):
+    mo.md(
+        r"""
+    ## Streaming Query
+
+    This query uses `SETTINGS stream = true` to continuously stream new blocks as they arrive.
+    The loader will automatically handle blockchain reorganizations.
+    """
+    )
+    return
+
+
+@app.cell
+def _(client):
+    streaming_results = client.sql(
+        '''
+        SELECT
+            block_num,
+            block_hash,
+            timestamp,
+            tx_hash,
+            log_index,
+            address,
+            topic0
+        FROM eth_firehose.logs
+        WHERE block_num >= 20000000
+        '''
+    ).load(
+        'my_kafka',
+        'eth_logs_stream',
+        stream=True,
+        create_table=True,
+    )
+    return (streaming_results,)
+
+
+@app.cell(hide_code=True)
+def _(mo):
+    mo.md(
+        r"""
+    ## Monitor Stream
+
+    This cell will continuously print results as they arrive. Press Ctrl+C to stop.
+    """
+    )
+    return
+
+
+@app.cell
+def _(streaming_results):
+    batch_count = 0
+    reorg_count = 0
+
+    for result in streaming_results:
+        if result.is_reorg:
+            reorg_count += 1
+            print(f'🔄 REORG #{reorg_count}: {len(result.invalidation_ranges)} ranges invalidated')
+            for r in result.invalidation_ranges:
+                print(f'   Network: {r.network}, Blocks: {r.start}-{r.end}')
+        else:
+            batch_count += 1
+            print(f'✅ Batch #{batch_count}: {result.rows_loaded} rows, {result.ops_per_second:.0f} rows/sec')
+
+            if 'block_ranges' in result.metadata:
+                for br in result.metadata['block_ranges']:
+                    print(f'   Blocks: {br["start"]}-{br["end"]} ({br["network"]})')
+    return
+
+
+@app.cell(hide_code=True)
+def _(mo):
+    mo.md(r"""## Verify Messages in Kafka""")
+    return
+
+
+@app.cell
+def _():
+    from kafka import KafkaConsumer
+    import json
+
+    consumer = KafkaConsumer(
+        'eth_logs_stream',
+        bootstrap_servers='localhost:9092',
+        auto_offset_reset='earliest',
+        consumer_timeout_ms=5000,
+        value_deserializer=lambda m: json.loads(m.decode('utf-8'))
+    )
+
+    messages = []
+    for message in consumer:
+        messages.append(message.value)
+        if len(messages) >= 10:
+            break
+
+    consumer.close()
+
+    print(f"Latest {len(messages)} messages:")
+    for i, msg in enumerate(messages):
+        print(f"\n[{i+1}] Block {msg.get('block_num')} at {msg.get('timestamp')}")
+        print(f"    TX: {msg.get('tx_hash')}")
+        print(f"    Address: {msg.get('address')}")
+    return
+
+
+
+@app.cell
+def _():
+    return
+
+
+if __name__ == "__main__":
+    app.run()