<jupyter_start><jupyter_text>TileDB Presto Connector Demo This notebook will demostrate the TileDB connector for Presto. Sample NYSE quote data is used for this demostration (see [demo prerequisites](https://github.com/TileDB-Inc/TileDB-Notebooks/blob/master/Presto_Demo/README.md))<jupyter_code>%matplotlib inline
import pandas as pd
import numpy as np
import sys
import prestodb<jupyter_output><empty_output><jupyter_text>Adjust the S3 bucket to the one you ingested the array in the prerequisites steps.<jupyter_code>s3_bucket = 's3://your_bucket_here/tiledb_arrays/'
quote_uri = s3_bucket + 'quote_20180730'<jupyter_output><empty_output><jupyter_text>Set the AWS access keys in order to access the array on the S3 bucket.<jupyter_code>AWS_ACCESS_KEY_ID=xxxx
AWS_SECRET_ACCESS_KEY=yyyy<jupyter_output><empty_output><jupyter_text>Connect to a local running Presto instance:<jupyter_code>conn=prestodb.dbapi.connect(
    host='127.0.0.1',
    port=8080,
    user='demo',
    catalog='tiledb',
    schema='tiledb',
    session_properties = {
        'tiledb.aws_access_key_id': AWS_ACCESS_KEY_ID,
        'tiledb.aws_secret_access_key': AWS_SECRET_ACCESS_KEY
    }
)
cur = conn.cursor()
cur.execute('SELECT * FROM system.runtime.nodes')
rows = cur.fetchall()
pd.DataFrame(rows, columns=("node_id", "http_uri", "node_version", "coordinator", "state"))<jupyter_output><empty_output><jupyter_text>Describe a TileDB array inside Presto. The TileDB schema is translated to Presto tabular format. Arrays can be accessed directly by URI, no pre-registration is needed inside Presto.<jupyter_code>cur.execute("""DESCRIBE tiledb.tiledb."{}"
""".format(quote_uri))
rows = cur.fetchall()
pd.DataFrame(rows, columns=("Column", "Type", "Extra", "Comment"))<jupyter_output><empty_output><jupyter_text>Just as with `DESCRIBE`, `SHOW CREATE TABLE` works.<jupyter_code>cur.execute("""SHOW CREATE TABLE tiledb.tiledb."{}"
""".format(quote_uri))
rows = cur.fetchall()
print(rows[0][0])<jupyter_output>CREATE TABLE tiledb.tiledb."s3://tiledb-presto/nyse/tiledb-arrays/quote/quote_20180730_symbol_datetime_sequence_gzip" (
   symbol_id bigint COMMENT 'Dimension',
   datetime bigint COMMENT 'Dimension',
   sequence_number bigint COMMENT 'Dimension',
   exchange varchar(1) COMMENT 'Attribute',
   symbol varchar COMMENT 'Attribute',
   bid_price real COMMENT 'Attribute',
   bid_size bigint COMMENT 'Attribute',
   offer_price real COMMENT 'Attribute',
   offer_size bigint COMMENT 'Attribute',
   quote_condition varchar(1) COMMENT 'Attribute',
   national_bbo_ind varchar(1) COMMENT 'Attribute',
   finra_bbo_indicator varchar(1) COMMENT 'Attribute',
   finra_adf_mpid_indicator smallint COMMENT 'Attribute',
   quote_cancel_correction varchar(1) COMMENT 'Attribute',
   source_of_quote varchar(1) COMMENT 'Attribute',
   retail_interest_indicator varchar(1) COMMENT 'Attribute',
   short_sale_restriction_indicator varchar(1) COMMENT 'Attribute',
   luld_bbo_indicator varchar(1) COMMENT 'Attribute'[...]<jupyter_text>Let's select some basic data. This will pull in 10 rows from the NYSE quote data for the Alphabet stock.<jupyter_code>cur.execute("""
SELECT symbol_id, datetime, symbol, bid_price, bid_size FROM tiledb.tiledb."{}"
WHERE datetime between 1532958300000000000 AND 1532958360000000000 AND symbol_id = 3377 limit 10
""".format(quote_uri))
rows = cur.fetchall()
pd.DataFrame(rows, columns=("symbol_id", "datetime", "symbol", "bid_price", "bid_size"))<jupyter_output><empty_output><jupyter_text>Let's run an aggregation query. This will select 6 hours of data for Alphabet stock, aggregate it to 15 minutes windows then find the first, last, min and max bid and offer prices.<jupyter_code>sql = """
WITH
quotes_w_vals AS
(select symbol,
symbol_id,
datetime,
datetime / (15*60*CAST(1000000000 AS bigint)) * (15*60*CAST(1000000000 AS bigint)) as fifteen_minute_timekey,
bid_price,
first_value(bid_price) OVER(PARTITION BY datetime / (15*60*CAST(1000000000 AS bigint)) * (15*60*CAST(1000000000 AS bigint)) ORDER BY datetime asc) first_bid_price,
last_value(bid_price) OVER(PARTITION BY datetime / (15*60*CAST(1000000000 AS bigint)) * (15*60*CAST(1000000000 AS bigint)) ORDER BY datetime asc) last_bid_price,
offer_price,
first_value(offer_price) OVER(PARTITION BY datetime / (15*60*CAST(1000000000 AS bigint)) * (15*60*CAST(1000000000 AS bigint)) ORDER BY datetime asc) first_offer_price,
last_value(offer_price) OVER(PARTITION BY datetime / (15*60*CAST(1000000000 AS bigint)) * (15*60*CAST(1000000000 AS bigint)) ORDER BY datetime asc) last_offer_price
FROM tiledb.tiledb."{}"
WHERE symbol_id = 3377 AND datetime between 1532966400000000000 AND 1532988000000000000)

SELECT symbol_id,
fifteen_minute_timekey,
from_unixtime(fifteen_minute_timekey / 1000000000) AS fifteen_minute_datetime,
min(bid_price) min_bid_price,
min(first_bid_price) first_bid_price,
max(last_bid_price) last_bid_price,
max(bid_price) as max_bid_price,
min(offer_price) min_offer_price,
min(first_offer_price) first_offer_price,
max(last_offer_price) last_offer_price,
max(offer_price) as max_offer_price,
count(*) quotes
FROM quotes_w_vals
GROUP BY symbol_id, fifteen_minute_timekey
ORDER BY fifteen_minute_timekey asc
""".format(quote_uri)
cur.execute(sql)
rows = cur.fetchall()
pd.DataFrame(rows, columns=("symbol_id", "fifteen_minute_timekey", "fifteen_minute_datetime", "min_bid_price", "first_bid_price", "last_bid_price", "max_bid_price", "min_offer_price", "first_offer_price", "last_offer_price", "max_offer_price", "quotes"))<jupyter_output><empty_output><jupyter_text>Storing Aggregate Results in a New Array Now let's create a new TileDB array and store the results of the above aggregate in this new array.<jupyter_code>agg_array_uri = s3_bucket + 'presto_demo_agg_results'
cur.execute('DROP TABLE IF EXISTS tiledb.tiledb.presto_demo_agg_results')
rows = cur.fetchall()

cur.execute("""
CREATE TABLE tiledb.tiledb.presto_demo_agg_results (
symbol_id bigint WITH (dimension=True, lower_bound = 0, upper_bound = 30000, extent = 100),
fifteen_minute_timekey bigint WITH (dimension=True, lower_bound = 0, upper_bound = 2000000000000000000, extent = 60000000000),
fifteen_minute_datetime varchar,
min_bid_price real,
first_bid_price real,
last_bid_price real,
max_bid_price real,
min_offer_price real,
first_offer_price real,
last_offer_price real,
max_offer_price real,
quotes bigint
) WITH (uri='{}', type='SPARSE')
""".format(agg_array_uri))
rows = cur.fetchall()

cur.execute("""
INSERT INTO tiledb.tiledb."{}"
("fifteen_minute_timekey", "symbol_id", "fifteen_minute_datetime", "min_bid_price", "first_bid_price", "last_bid_price", "max_bid_price", "min_offer_price", "first_offer_price", "last_offer_price", "max_offer_price", "quotes")
WITH
quotes_w_vals AS
(select symbol,
symbol_id,
datetime,
datetime / (15*60*CAST(1000000000 AS bigint)) * (15*60*CAST(1000000000 AS bigint)) as fifteen_minute_timekey,
bid_price,
first_value(bid_price) OVER(PARTITION BY datetime / (15*60*CAST(1000000000 AS bigint)) * (15*60*CAST(1000000000 AS bigint)) ORDER BY datetime asc) first_bid_price,
last_value(bid_price) OVER(PARTITION BY datetime / (15*60*CAST(1000000000 AS bigint)) * (15*60*CAST(1000000000 AS bigint)) ORDER BY datetime asc) last_bid_price,
offer_price,
first_value(offer_price) OVER(PARTITION BY datetime / (15*60*CAST(1000000000 AS bigint)) * (15*60*CAST(1000000000 AS bigint)) ORDER BY datetime asc) first_offer_price,
last_value(offer_price) OVER(PARTITION BY datetime / (15*60*CAST(1000000000 AS bigint)) * (15*60*CAST(1000000000 AS bigint)) ORDER BY datetime asc) last_offer_price
FROM tiledb.tiledb."{}"
WHERE symbol_id = 3377 AND datetime between 1532966400000000000 AND 1532988000000000000)


SELECT
symbol_id,
fifteen_minute_timekey,
CAST(from_unixtime(fifteen_minute_timekey / 1000000000) AS varchar) AS fifteen_minute_datetime,
min(bid_price) min_bid_price,
min(first_bid_price) first_bid_price,
max(last_bid_price) last_bid_price,
max(bid_price) as max_bid_price,
min(offer_price) min_offer_price,
min(first_offer_price) first_offer_price,
max(last_offer_price) last_offer_price,
max(offer_price) as max_offer_price,
count(*) quotes
FROM quotes_w_vals
GROUP BY symbol_id, fifteen_minute_timekey
ORDER BY fifteen_minute_timekey asc
""".format(agg_array_uri, quote_uri))
rows = cur.fetchall()

cur.execute("""
SELECT *
FROM tiledb.tiledb."{}"
""".format(agg_array_uri))
rows = cur.fetchall()
pd.DataFrame(rows, columns=("symbol_id", "fifteen_minute_timekey", "fifteen_minute_datetime", "min_bid_price", "first_bid_price", "last_bid_price", "max_bid_price", "min_offer_price", "first_offer_price", "last_offer_price", "max_offer_price", "quotes"))<jupyter_output><empty_output><jupyter_text>Querying Presto-Created TileDB Arrays Directly from Python We've demostrated selecting and creating tables via Presto, let's now show that the exact same TileDB array can be accessed via python using the [TileDB-Py](https://github.com/TileDB-Inc/TileDB-Py) package.<jupyter_code>import tiledb
config = tiledb.Config()
config.set("aw_access_key_id", AWS_ACCESS_KEY_ID)
config.set("aws_secret_access_key", AWS_SECRET_ACCESS_KEY)
ctx = tiledb.Ctx(config)
schema = tiledb.ArraySchema.load(ctx, agg_array_uri)
domain = schema.domain
pd.DataFrame([(domain.dim(i).name, domain.dim(i).dtype) for i in range(domain.ndim)], columns=("Dimension", "dtype"))
with tiledb.SparseArray(ctx, agg_array_uri, mode='r') as array:
    res = array.query(
        attrs=("min_bid_price", "first_bid_price", "last_bid_price", "max_bid_price", "min_offer_price", "first_offer_price", "last_offer_price", "max_offer_price",))[:, :]

res['symbol_id'] = np.ascontiguousarray(res['coords']['symbol_id'])
res['fifteen_minute_timekey'] = np.ascontiguousarray(res['coords']['fifteen_minute_timekey'])
res['fifteen_minute_timekey_datetime'] = pd.to_datetime(res['fifteen_minute_timekey'])

df = pd.DataFrame(res)
df
df.drop(['fifteen_minute_timekey', 'symbol_id', 'coords'], axis=1).plot(x='fifteen_minute_timekey_datetime', title='Alphabet Stock Aggregate 2018/07/30', figsize=(20, 20))<jupyter_output><empty_output><jupyter_text>Performing Aggregation via TileDB-Py instead of Presto Here we will query the array and perform the same aggregation we did in Presto. One benefit of TileDB is its iteroperability. Data can be accessed directly from the TileDB Python API (or other language bindings) or through a SQL layer such as Presto. You are more comfortable with.<jupyter_code>schema = tiledb.ArraySchema.load(ctx, quote_uri)
domain = schema.domain
pd.DataFrame([(domain.dim(i).name, domain.dim(i).dtype) for i in range(domain.ndim)], columns=("Dimension", "dtype"))
with tiledb.SparseArray(ctx, quote_uri, mode='r') as array:
    # Query to get 6 hours of data, datetime is stored as unix epoch nanoseconds, we use pandas to query via human readable string
    starttime = pd.to_datetime('2018-07-30 16:00:00').to_datetime64().astype('int64')
    endtime = pd.to_datetime('2018-07-30 22:00:00').to_datetime64().astype('int64')
    print("Querying with starttime={}, endtime={}".format(starttime, endtime))
    res = array.query(
        attrs=("Bid_Price", "Bid_Size", "Offer_Price", "Offer_Size"))[3377:3378, starttime:endtime, :]
res['symbol_id'] = np.ascontiguousarray(res['coords']['symbol_id'])
res['epoch_datetime'] = np.ascontiguousarray(res['coords']['datetime'])
res['Sequence_Number'] = np.ascontiguousarray(res['coords']['Sequence_Number'])
res['datetime'] = pd.to_datetime(res['epoch_datetime'])
res['fifteen_minute_timekey'] = res['epoch_datetime'] // (15*60*1000000000) * (15*60*1000000000)
df = pd.DataFrame(res)
df.count()
df[0:10]<jupyter_output><empty_output><jupyter_text>Next we will perform the same operation as above via Pandas `groupby` functionality.<jupyter_code>df.groupby(['symbol_id', 'fifteen_minute_timekey']).agg({'fifteen_minute_timekey': lambda x: pd.to_datetime(x.min()), 'epoch_datetime': ['min', lambda x: pd.to_datetime(x.min())], 'Bid_Price': ['first', 'last', 'min', 'max'], 'Offer_Price': ['first', 'last', 'min', 'max']})<jupyter_output><empty_output><jupyter_text>There are a lot of 0's for minimum values in this aggregation, let's take a look at a plot of the raw data.<jupyter_code>df.drop(['symbol_id', 'epoch_datetime', 'sequence_number', 'Offer_Size', 'Bid_Size', 'fifteen_minute_timekey'], axis=1).plot(x='datetime', title='6 Hour Quotes for Alphabet Stock', figsize=(18, 16))<jupyter_output><empty_output><jupyter_text>Removing Outliers with Pandas The raw data graph shows the volume of 0 quotes, this happens where there is a bid with no offer, or an offer with no bid. Let's remove outliers via inter-quartile range method.<jupyter_code># Compute Offier Price IQR
OPQ1 = df['Offer_Price'].quantile(0.25)
OPQ3 = df['Offer_Price'].quantile(0.75)
OPIQR = OPQ3 - OPQ1
OP_lower = OPQ1 - 1.5 * OPIQR
OP_upper = OPQ3 + 1.5 * OPIQR
print("Offer_Price: Q1={}, Q3={}, IQR={}, lower={}, upper={}".format(OPQ1, OPQ3, OPIQR, OP_lower, OP_upper))

# Compute Bid Price IQR
BPQ1 = df['Bid_Price'].quantile(0.25)
BPQ3 = df['Bid_Price'].quantile(0.75)
BPIQR = BPQ3 - BPQ1
BP_lower = BPQ1 - 1.5 * BPIQR
BP_upper = BPQ3 + 1.5 * BPIQR
print("Bid_Price: Q1={}, Q3={}, IQR={}, lower={}, upper={}".format(BPQ1, BPQ3, BPIQR, BP_lower, BP_upper))

# Filtering Values to those that are inside the Bid/Offer price range
filtered = df.query('@OP_lower <= Offer_Price <= @OP_upper & @BP_lower <= Bid_Price <= @BP_upper')
filtered.drop(['symbol_id', 'epoch_datetime', 'sequence_number', 'Offer_Size', 'Bid_Size', 'fifteen_minute_timekey'], axis=1).plot(x='datetime', title='6 Hour Quotes for Alphabet Stock (Filtered)', figsize=(18, 16))<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.12600791672775252, 'fraction_numerical': 0.04838000293212139, 'mean_word_length': 7.607570977917981, 'pattern_counts': {'":': 0, '<': 60, '<?xml version=': 0, '>': 56, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 1, 'substrings_counts': 32, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '14', 'dir': 'jupyter-structured-clean-dedup', 'id': '622175', 'license': 'MIT', 'n_tokens_mistral': 5305, 'n_tokens_neox': 4746, 'n_words': 1112, 'path': 'presto_demo/TileDB_Presto_Demo.ipynb', 'repo_name': 'TileDB-Inc/TileDB-Notebooks', 'size': '13642'}