<jupyter_start><jupyter_text>Elasticsearch Data AnalyticsThis notebook provides sample code to fetch Elasticsearch Data into and analyze it.<jupyter_code>from IPython.display import Image
from IPython.core.display import HTML 
Image(url= "https://www.antaresnet.com/wp-content/uploads/2018/07/Elasticsearch-Logo-Color-V.png")<jupyter_output><empty_output><jupyter_text>Loading modules and connect to the Elastic Stack<jupyter_code>import sys
import json
import requests
import numpy as np
import pandas as pd
import seaborn as sns
import pytz
from datetime import datetime, timedelta
from dateutil import tz

import matplotlib.pyplot as plt
sns.set(style="darkgrid")
plt.rcParams["figure.figsize"] = (18,10)
# connect to our cluster
from elasticsearch import Elasticsearch
es = Elasticsearch([{'host': 'elasticsearch', 'port': 9200}])
print('Last run: {} UTC, status: {} %'.format(
    datetime.utcnow(),
    es.cluster.health()['active_shards_percent_as_number']))
es.info()<jupyter_output>Last run: 2019-08-07 14:37:15.946994 UTC, status: 80.0 %<jupyter_text>Display our indices and document types saved in elasticsearch.  Defining useful functions<jupyter_code>!pip install "elasticsearch>=7.0.0,<8.0.0"
# %load scroller.py
def scroller(index, quantity, timerange=timedelta(days=0), startdt="", enddt=""):
    print("Starting to scroll", end='')
    # Retrieve the datetimes, note that timerange has a higher priority
    if timerange.total_seconds() > 0:
        now = datetime.utcnow().replace(tzinfo=pytz.UTC)
        startdt = (now - timerange).isoformat()
        enddt = now.isoformat()
    
    # search the first page and write the result to data
    response = es.search(
        index=index,
        body={
                      "query": {
                        "bool": {
                          "must": [
                            {"range" : {
                                "phenomenonTime" : {
                                    #"gte": "2018-02-20T09:08:34.230693+00:00", 
                                    "gte": startdt,
                                    "lte": enddt, 
                                    "time_zone": "+01:00"
                                }
                            }},
                            {
                              "match_phrase": {
                                "Datastream.name.keyword": quantity
                              }
                            }
                          ]
                        }
                      }
                    },
        scroll='10m'
    )
    data = [[row["_source"]["phenomenonTime"], row["_source"]["result"]] for row in response['hits']['hits']]

    # Append new pages until there aren't any left
    while len(response['hits']['hits']):
        print(".", end='')
        # process results
        # print([item["_id"] for item in response["hits"]["hits"]])
        response = es.scroll(scroll_id=response['_scroll_id'], scroll='10m')
        data +=  [[row["_source"]["phenomenonTime"], row["_source"]["result"]] for row in response['hits']['hits']]
    
    # Convert data to a DataFrame and return it
    df = pd.DataFrame(data, columns=["phenomenonTime", quantity])
    # df.index = pd.to_datetime(df["phenomenonTime"].map(lambda t: t.split(".")[0]), utc=True)
    df.index = pd.to_datetime(df["phenomenonTime"].map(lambda t: roundto(t, 1)), utc=True)
    df = df.drop(["phenomenonTime"], axis=1)
    print("\nFetched {} tuples.".format(df.shape[0]))
    return df

def roundto(string, n):
    base = string.split(".")[0]
    if n > 0:
        base += "." + string.split(".")[1][:n]
    return base<jupyter_output><empty_output><jupyter_text>Gather data from Elasticsearch<jupyter_code># Get data for an index and a quantity between two static timestamps
startdt="2019-08-07T08:58:34+00:00"
enddt="2019-08-07T11:58:34+00:00"
df = scroller("at.srfg.iot-iot4cps-wp5.infraprov.internal-*",
              "at.srfg.iot-iot4cps-wp5.CarFleet1.car_1.Air Temperature",
              startdt=startdt, enddt=enddt)
df.head()
# Get data for an index and a quantity of the latest timerange
df = scroller("at.srfg.iot-iot4cps-wp5.infraprov.internal-*",
              "at.srfg.iot-iot4cps-wp5.CarFleet1.car_1.Air Temperature",
              timerange=timedelta(days=10))
df.head()
# Plot the extracted data using pandas and seaborn
df.plot()
plt.show()
# Get multiple quantities and (outer) join them to a single DataFrame.
# There can be a lot of missing values
used_quantities = ["at.srfg.iot-iot4cps-wp5.CarFleet1.car_1.Air Temperature", 
                   "at.srfg.iot-iot4cps-wp5.CarFleet2.car_2.Air Temperature"]
df = scroller("at.srfg.iot-iot4cps-wp5.infraprov.internal-*",
              used_quantities[0],
              timerange=timedelta(days=10))
for q in used_quantities[1:]:
    print(q)
    df = df.join(scroller("at.srfg.iot-iot4cps-wp5.infraprov.internal-*", q,
                          timerange=timedelta(days=10)),
                 how="outer")
df.head()<jupyter_output><empty_output><jupyter_text>Store and retrieve the DataFrame in a csv<jupyter_code>df.to_csv("elasticsearchdata.csv")
df = pd.read_csv("elasticsearchdata.csv", parse_dates=True, index_col='phenomenonTime')
df.tail()<jupyter_output><empty_output><jupyter_text>Pre-processing  Reduce size and interpolate<jupyter_code>df = pd.read_csv("elasticsearchdata.csv", parse_dates=True, index_col='phenomenonTime')
df.index.names = ["time"]
col_mapping = {"at.srfg.iot-iot4cps-wp5.CarFleet1.car_1.Air Temperature": "car1_temp", 
                        "at.srfg.iot-iot4cps-wp5.CarFleet2.car_2.Air Temperature": "car2_temp"}
df = df.rename(index=str, 
               columns=col_mapping)
df.head()
# Interpolate forwards and backwaonly up to 
df = df.interpolate(method ='linear', limit_direction ='both', limit=10)
df = df.interpolate(method ='linear', limit_direction ='both', limit=10)
# Keep only the rows with at least 2 non-NA values.
df = df.dropna(thresh=2)
# make Timestamp unique
df = df.reset_index()
df = df.groupby("time").agg({q: "mean" for q in col_mapping.values()})
# Interpolate again to close gaps, use the smalles value 
df = df.interpolate(method ='zero', limit_direction ='forward')
df = df.interpolate(method ='zero', limit_direction ='forward')
df.index.min()
df.shape
df.isna().sum()
df.describe()
# Keep only rows with all filled rows
df = df.dropna()
df.to_csv("elasticsearchdata.csv")<jupyter_output><empty_output><jupyter_text>Basic Data Analysis<jupyter_code>df = pd.read_csv("elasticsearchdata.csv", parse_dates=True, index_col='time')
df.tail()
# df.hist()
plt.show()
# pd.plotting.scatter_matrix(df, alpha=0.2)
plt.show()
# corr = df.corr() 
cm = sns.light_palette("orange", as_cmap=True) 
cm = sns.diverging_palette(220, 20, sep=20, as_cmap=True) 
# corr.style.background_gradient(cmap=cm).set_precision(2)<jupyter_output><empty_output><jupyter_text>Feature EngineeringThis task is very domain-specific and must be done by an expert.  Data Analytics<jupyter_code>from IPython.html.widgets import *
from mpl_toolkits.mplot3d import Axes3D

plt.rcParams["figure.figsize"] = (18,10)
sns.set(style="darkgrid")

def plot3D(pitch, yaw):
    fig = plt.figure()
    ax = fig.add_subplot(111, projection='3d')
    plot = ax.scatter(df['car1_temp'], df['car1_temp'], df['car2_temp'], c=df["car1_temp"], s=60)
    fig.colorbar(plot)
    ax.view_init(pitch, yaw)
    ax.legend(['Vibration for each 3D position'])
    ax.set_xlabel("x-Position")
    ax.set_ylabel("y-Position")
    ax.set_zlabel("z-Position")
interact(plot3D, pitch=(0,90,1), yaw=(0,90,1))
plt.show()
# df[col_mapping.values()].hist()
plt.show()
# pd.plotting.scatter_matrix(df[["vib", "distance", "projection", "v-radial", "v-tang"]], alpha=0.5)
plt.show()
# bins = np.linspace(0, df['v-radial'].max(), 10)
# df["binned-v-radial"] = pd.cut(df['v-radial'], bins)
# df.groupby("binned-v-radial").agg({"vib": {"min", "median", "mean", "max", "count"}})
# corr = df[df.columns.sort_values()].corr()[["vib", "vib-x", "vib-y"]]
# cm = sns.light_palette("orange", as_cmap=True) 
# cm = sns.diverging_palette(220, 20, sep=20, as_cmap=True) 
# corr.style.background_gradient(cmap=cm).set_precision(2)
from IPython.display import Markdown 
Markdown("""
## Inline Description:

It is very nice to describe results using Markdown with dynamic values like: **{pi}**.

""".format(pi=np.pi))<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.14990467111534794, 'fraction_numerical': 0.025977121067683507, 'mean_word_length': 3.2908997955010224, 'pattern_counts': {'":': 12, '<': 33, '<?xml version=': 0, '>': 35, 'https://': 1, 'lorem ipsum': 0, 'www.': 1, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '8', 'dir': 'jupyter-structured-clean-dedup', 'id': '56574', 'license': 'Apache-2.0', 'n_tokens_mistral': 3039, 'n_tokens_neox': 2883, 'n_words': 577, 'path': 'demo_applications/Analytics/jupyter/src/provisioning/data/Getting_Started/Elastic Connection.ipynb', 'repo_name': 'iot-salzburg/panta-rhei', 'size': '8392'}