Python SDK

from risingwave import RisingWave, RisingWaveConnOptions  # Connect to RisingWave instance on localhost with named parameters rw = RisingWave(  RisingWaveConnOptions.from_connection_info(  host="localhost", port=4566, user="root", password="root", database="dev"  ) )  # Connect to RisingWave instance on localhost with connection string rw = RisingWave(RisingWaveConnOptions("postgresql://root:root@localhost:4566/dev"))  # You can create a new SQL connection and execute operations under the with statement.  # This is the recommended way for python sdk usage. with rw.getconn() as conn:  conn.insert(...)  conn.fetch(...)  conn.execute(...)  conn.mv(...)  conn.on_change(...)   # You can also use the existing connection created by RisingWave object to execute operations. # This will be used in the later sections for simplicity. rw.insert(...) rw.fetch(...) rw.execute(...) rw.mv(...) rw.on_change(...)  

from datetime import datetime import pandas as pd  df = pd.DataFrame(  {  "product": ["foo", "bar"],  "price": [123.4, 456.7],  "ts": [datetime.strptime("2023-10-05 14:30:00", "%Y-%m-%d %H:%M:%S"),   datetime.strptime("2023-10-05 14:31:20", "%Y-%m-%d %H:%M:%S")],  } )  # A test table will be created if not exist in risingwave with the correct schema rw.insert(table_name="test", data=df)  # You can provide an optional force_flush parameter and set it to True # if you would the inserted data to be visible in fetch query immediately. # Otherwise, data will be inserted in batches asynchronously for better performance. # rw.insert(table_name="test", data=df, force_flush = True) 

# Create a table and load data from upstream kafka rw.execute("""  CREATE TABLE IF NOT EXISTS source_abc  WITH (  connector='kafka',  properties.bootstrap.server='localhost:9092',  topic='test_topic'  )   FORMAT UPSERT ENCODE AVRO (  schema.registry = 'http://127.0.0.1:8081',  schema.registry.username='your_schema_registry_username',  schema.registry.password='your_schema_registry_password' )""") 

from risingwave import OutputFormat  result: pd.DataFrame = rw.fetch("""  SELECT window_start, window_end, product, ROUND(avg(price)) as avg_price  FROM tumble(test, ts, interval '10 seconds')   GROUP BY window_start, window_end, product""",   format=OutputFormat.DATAFRAME)  print(result) # Output: # window_start window_end product avg_price # 0 2023-10-05 14:31:20 2023-10-05 14:31:30 bar 457.0 # 1 2023-10-05 14:30:00 2023-10-05 14:30:10 foo 123.0  # You can also use OutputFormat.RAW to get back list of tuples as the query results # rw.fetch("...", format=OutputFormat.RAW) # [(datetime.datetime(2023, 10, 5, 14, 31, 20), datetime.datetime(2023, 10, 5, 14, 31, 30), 'bar', 457.0),  # (datetime.datetime(2023, 10, 5, 14, 30), datetime.datetime(2023, 10, 5, 14, 30, 10), 'foo', 123.0)] 

mv = rw.mv(name="test_mv",  stmt="""SELECT window_start, window_end, product, ROUND(avg(price)) as avg_price  FROM tumble(test, ts, interval '10 seconds')   GROUP BY window_start, window_end, product""") 

# Write your own handler # the event will contains all fields of the subscribed MV/Table plus two additional columns: # - op: varchar. The change operations. Valid values: [Insert, UpdateInsert, Delete, UpdateDelete].  # The reason why we have UpdateInsert and UpdateDelete is because RisingWave treats an UPDATE  # as a delete of the old value followed by an insert of the new value. # - rw_timestamp: bigint. The Unix timestamp in milliseconds when the data was written. def simple_event_handler(event: pd.DataFrame):  for _, row in event.iterrows():  # Trigger an action (e.g. place an order via REST API) when the avg_price exceeds 300  if (row["op"] == "UpdateInsert" or row["op"] == "Insert") and row["avg_price"] >= 300:  print(  f"{row['window_start']} - {row['window_end']}: {row['product']} avg price {row['avg_price']} exceeds 300")  # ...   import threading  # Subscribe changes of a materialized view and feed it to your own handler. # Run on_change in a separate thread without blocking the main thread. threading.Thread(  target=lambda: mv.on_change(  # Pass your handler here  handler = simple_event_handler,  # Support DATAFRAME and RAW tuples here  output_format=OutputFormat.DATAFRAME,  # If set to True, progress of the subscription will be saved and can be recovered on python application crashed  persist_progress=True,   # Maximal number of rows returned each time  max_batch_size = 10)  ).start()  # Subscribe changes of a table and print them to console. # Run on_change in a separate thread without blocking the main thread. threading.Thread(  target=lambda: rw.on_change(  subscribe_from="test",  handler = lambda data: print(data),  output_format=OutputFormat.RAW,  persist_progress=False,   max_batch_size = 5)  ).start()   # Future inserted data into the base table will be reflected in the subscriptions # df = pd.DataFrame( # { # "product": ["foo", "bar"], # "price": [1000.2, 5000.4], # "ts": [datetime.strptime("2023-10-05 17:30:00", "%Y-%m-%d %H:%M:%S"),  # datetime.strptime("2023-10-05 17:31:20", "%Y-%m-%d %H:%M:%S")], # } # ) # rw.insert(table_name="test", data=df)