# ---
# jupyter:
#   jupytext:
#     text_representation:
#       extension: .py
#       format_name: light
#       format_version: '1.5'
#       jupytext_version: 1.14.4
#   kernelspec:
#     display_name: Python 3
#     language: python
#     name: python3
# ---

# # Explore Real-Time Data on Iguazio via Integrated Notebooks

# ## Analyze Real-Time Data Using Spark Streaming, SQL, and ML
# Data apears to Spark as native Spark Data Frames

from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("Iguazio Integration demo").getOrCreate()

spark.read.format("io.iguaz.v3io.spark.sql.kv").load("v3io://bigdata/stocks").show()

# ## Run Interactive SQL Queries on Real-Time Data
# Support full ANSI SQL through native integration of Apache Presto over iguazio DB

# %sql select * from bigdata.stocks where exchange='NASDAQ'

# ## Read NoSQL Data as Real-Time Data Frame Stream

kvdf = pd.concat(client.read(backend='kv', table='stocks', filter='exchange=="NASDAQ"'), sort=False)
kvdf.head(8)

# ## Read the Time-Series data (in iguazio TSDB) as Pandas Data Frame
# Use DB side aggregations, joins and filtering on the real-time metrics data

import pandas as pd
import v3io_frames as v3f
client = v3f.Client('http://v3io-framesd:8080')

# +
# Read Time-Series aggregates from the DB (returned as a data stream, use concat to assemble the frames)
dfs = client.read(backend='tsdb', table='stock_metrics', step='60m', aggragators='avg,max,min',start="now-2d", end='now')
merged = pd.concat(dfs, sort=False)

# turn the results to a Multi-indexed and pivoted table 
pvt = merged.pivot_table('values',['Date','symbol','exchange'],['metric_name','Aggregate'])
pvt.head()
# -

# ## Run interesting Analysis On Real-Time Data
# e.g. compare stock price volatility between 2nd tier cloud providers

import matplotlib.pyplot as plt
# %matplotlib inline

# +
stock_symbols = ['GOOG','MSFT','ORCL','IBM']   
fig, axarr = plt.subplots(1,4)
pricesdf = merged.loc[merged['metric_name']=='price']

for sym, ax in zip(stock_symbols, axarr):
    pricesdf.loc[pricesdf['symbol'] == sym].pivot(columns='Aggregate', values='values').plot(ax=ax, title=sym, figsize=[20,5])
# -

# ## Read the Twitter + Sentiments feed as Pandas Data Frame
# Streaming data generated in by nuclio functions can be read in real-time or historically. <br>It can be distributed to multiple workers for scalability via sharding/partitioning. 

streamdfs = client.read(backend='stream', table='stocks_stream',seek='seq',shard='0', sequence=140)
streamdf = pd.concat(streamdfs, sort=False)
streamdf.head()

# ## Save Any Data "To Go" as a CSV file (or other formats)

streamdf.to_csv('mystream.csv')

StarcoderdataJupyterScriptsDedupFiltered