# -*- coding: utf-8 -*-
# ---
# jupyter:
#   jupytext:
#     text_representation:
#       extension: .py
#       format_name: light
#       format_version: '1.5'
#       jupytext_version: 1.14.4
#   kernelspec:
#     display_name: Python 3.6 with Spark
#     language: python3
#     name: python36
# ---

# # Querying IBM Db2 Event Store
# IBM Db2 Event Store is a hybrid transactional/analytical processing (HTAP) system. This notebook will demonstrate the best practices for querying a table stored in IBM Db2 Event Store.
#
# ***Pre-Req: Event_Store_Table_Creation***

# ## Connect to IBM Db2 Event Store
#
# Edit the values in the next cell

# +
CONNECTION_ENDPOINT=""

EVENT_USER_ID=""

EVENT_PASSWORD=""

# Port will be 1100 for version 1.1.2 or later (5555 for version 1.1.1)
PORT = "30370"

DEPLOYMENT_ID=""

# Database name
DB_NAME = "EVENTDB"

# Table name
TABLE_NAME = "IOT_TEMPERATURE"

HOSTNAME=""

DEPLOYMENT_SPACE=""
# -

# ## Import Python modules

# +
## Note: Only run this cell if your IBM Db2 Event Store is installed with IBM Cloud Pak for Data (CP4D)

# In IBM Cloud Pak for Data, we need to create link to ensure Event Store Python library is 
# properly exposed to the Spark runtime.
from pathlib import Path
src = '/home/spark/user_home/eventstore/eventstore'
dst = '/home/spark/shared/user-libs/python3.6/eventstore'
is_symlink = Path(dst).is_symlink()
if is_symlink == False :
    os.symlink(src, dst)
    print("Creating symlink to include Event Store Python library...")
else:
    print("Symlink already exists, not creating..")

# -

bearerToken=!echo `curl --silent -k -X GET https://{HOSTNAME}:443/v1/preauth/validateAuth -u {EVENT_USER_ID}:{EVENT_PASSWORD} |python -c "import sys, json; print(json.load(sys.stdin)['accessToken'])"`
bearerToken=bearerToken[0]
keystorePassword=!echo `curl -k --silent  GET -H "authorization: Bearer {bearerToken}" "https://{HOSTNAME}:443/icp4data-databases/{DEPLOYMENT_ID}/zen/com/ibm/event/api/v1/oltp/keystore_password"`
keystorePassword

from eventstore.common import ConfigurationReader
from eventstore.oltp import EventContext
from eventstore.sql import EventSession
from pyspark.sql import SparkSession

# ## Connect to Event Store

ConfigurationReader.setConnectionEndpoints(CONNECTION_ENDPOINT)
ConfigurationReader.setEventUser(EVENT_USER_ID)
ConfigurationReader.setEventPassword(<PASSWORD>_PASSWORD)
ConfigurationReader.setSslKeyAndTrustStorePasswords(keystorePassword[0])
ConfigurationReader.setDeploymentID(DEPLOYMENT_ID)
ConfigurationReader.getSslTrustStorePassword()


# ## Open the database
#
# The cells in this section are used to open the database and create a temporary view for the table that we created previously.   

# To run Spark SQL queries, you first have to set up a Db2 Event Store Spark session. The EventSession class extends the optimizer of the SparkSession class.

sparkSession = SparkSession.builder.appName("EventStore SQL in Python").getOrCreate()
eventSession = EventSession(sparkSession.sparkContext, DB_NAME)

# The next cell opens the database to allow operations against it to be executed.

eventSession.open_database()

# With the following cells, we can list all existing tables and then load the table we previously created into the tab DataFrame reference. Note that we are defining the `tab` DataFrame reference that will be used later on in this notebook to create a temporary view.

# +
with EventContext.get_event_context(DB_NAME) as ctx:
   print("Event context successfully retrieved.")

print("Table names:")
table_names = ctx.get_names_of_tables()
for name in table_names:
   print(name)
# -

tab = eventSession.load_event_table(TABLE_NAME)

# Let's recall the table schema we previously created.

try:
    resolved_table_schema = ctx.get_table(TABLE_NAME)
    print(resolved_table_schema)
except Exception as err:
    print("Table not found")

# ## Best Practices for efficient queries

# In the next cell we create a lazily evaluated "view" that we can then use like a hive table in Spark SQL, but this is only evaluated when we actually run or cache query results. We are calling this view "readings" and that is how we will refer to it in the queries below:

tab.createOrReplaceTempView("readings")

query = "SELECT count(*) FROM readings"
print("{}\nRunning query in Event Store...".format(query))
df_data = eventSession.sql(query)
df_data.toPandas()

# Let's have a look at the record structure 

query = "SELECT * FROM readings LIMIT 1"
print("{}\nRunning query in Event Store...".format(query))
df_data = eventSession.sql(query)
df_data.toPandas()

query = "SELECT MIN(ts), MAX(ts) FROM readings"
print("{}\nRunning query in Event Store...".format(query))
df_data = eventSession.sql(query)
df_data.toPandas()


# ## Optimal query through the index
#
# - Index queries will significantly reduce amount of data that needs to be scanned for results. 
#     - Indexes in IBM Db2 Event Store are formed asynchronously to avoid insert latency. 
#     - They are stored as a Log Structured Merge (LSM) Tree.
#     - The index is formed by "runs", which include sequences of sorted keys. These runs are written to disk during “Share” processing.
#     - These index runs are merged together over time to improve scan and I/O efficiency.
#
# - For an optimal query performance you must specify equality on all the equal_columns in the index and a range on the sort column in the index.

# For example, in the following query we are retrieving all the values in the range of dates for a specific device and sensor, where both the `deviceID` and `sensorID` are in the equal_columns definition for the index schema and the `ts` column is the sort column for the index.

# Then the following cell runs the query and caches the results. Note that this caching is for demostration purposes, to show the time it takes to run the query and cache the results in memory within Spark. This caching is recommended when you are going to do additional processing on this cached data, as the query against IBM Db2 Event Store is only run once. 

%%time
index_query = "SELECT ts, temperature  FROM readings where deviceID=1 and sensorID=12 and ts >1541021271619 and ts < 1541043671128 order by ts"
print("{}\nCreating a dataframe for the query ...".format(index_query))
df_index_query = eventSession.sql(index_query)
df_index_query.cache()

# Finally the results from the cached data can be visualized:

df_index_query.toPandas()

# ## Sub-optimal query
#
# This next query shows a sub-optimal query that only specifies equality in one of the equal_columns in the index schema, and for this reason ends up doing a full scan of the table.

%%time
fullscan_query = "SELECT count(*) FROM readings where sensorID = 7"
print("{}\nCreating a dataframe for the query...".format(fullscan_query))
df_fullscan_query = eventSession.sql(fullscan_query)
df_fullscan_query.cache()

df_fullscan_query.toPandas()

# ## Accessing multiple sensorIDs optimally
#
# The easiest way to write a query that needs to retrieve multiple values in the equal_columns in the index schema is by using an *In-List*. With this, you can get optimal index access across multiple sensorID's. 
#
# In this example we specify equality for a specific deviceID, and an In-List for the four sensors we are trying to retrieve. To limit the number of records we are returning we also include a range of timestamps.

%%time
inlist_query = "SELECT deviceID, sensorID, ts FROM readings where deviceID=1 and sensorID in (1, 5, 7, 12) and ts >1541021271619 and ts < 1541043671128 order by ts"
print("{}\nCreating a dataframe for the query...".format(inlist_query))
df_inlist_query = eventSession.sql(inlist_query)
df_inlist_query.cache()

df_inlist_query.toPandas()

# ## Exploiting the synopsis table:  Advanced medium weight queries
#
# - Event Store tables include a synopsis table which summarizes the minimum/maximum values of the data for each range of rows in the synopsis table. 
#
#      - It contains one range for every 1000 rows.
#      - It is stored in a separate internal table in the shared storage layer.
#      - It is parquet compressed to minimize footprint.
#      - For highly selective queries, it can improve performance by up to 1000x.
#
# - Using an equality or a range predicate on a clustered field (e.g. the 'ts' column in our case because the data is inserted into the table in order) is faster than doing a full scan as it should be able to exploit the synopsis table, but this will be slower than an optimal index scan.

%%time
synopsis_query = "SELECT deviceID, sensorID, ts FROM readings where ts >1541021271619 and ts < 1541043671128 order by ts"
print("{}\nCreating a dataframe for the query...".format(synopsis_query))
df_synopsis_query = eventSession.sql(synopsis_query)
df_synopsis_query.cache()

df_synopsis_query.toPandas()

# ## Summary
# This demo introduced you to the best practices querying the table stored in IBM Db2 Event Store database.
#
# ## Next Step
# `"Event_Store_Data_Analytics.ipynb"` will show you how to perform data analytics with IBM Db2 Event Store with multiple scientific tools.

# <p><font size=-1 color=gray>
# &copy; Copyright 2019 IBM Corp. All Rights Reserved.
# <p>
# Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file
# except in compliance with the License. You may obtain a copy of the License at
# https://www.apache.org/licenses/LICENSE-2.0
# Unless required by applicable law or agreed to in writing, software distributed under the
# License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
# express or implied. See the License for the specific language governing permissions and
# limitations under the License.
# </font></p>



StarcoderdataJupyterScriptsDedupFiltered