dask-px-search-splunk.py

from dask.distributed import Client, progress
client = Client(threads_per_worker=8, n_workers=2)
client

import urllib
import requests
import aiohttp
from pprint import pprint
import dask
import dask.dataframe as dd
import pandas as pd
from io import StringIO
import io, pkgutil
import csv

client.cluster.scale(4) 

# disable SSL warnings
import urllib3
urllib3.disable_warnings()
#df = dd.read_csv(filename)

baseurl = 'https://localhost:8089'
userName = 'admin'
password = 'welcome1'
url = baseurl + '/servicesNS/admin/search/search/jobs/export'
import dask.dataframe as dd

results_dd = dd

@dask.delayed
def get_results(number_of_bins,filter_bin_num):

    #searchQuery = 'search source="2019-01-flights-full.csv.gz" index="flights" '+ \    
    searchQuery = 'search index="flights" '+ \
        ' | eval bin_num=tonumber(substr(_cd,-3))%'+str(number_of_bins) + \
        ' | where bin_num = '+str(filter_bin_num) + \
        ' | table _time,ARR_DELAY,ARR_TIME,CANCELLED,DEP_DELAY_NEW,DEP_TIME,DEST,DEST_AIRPORT_ID,DEST_AIRPORT_SEQ_ID,DEST_CITY_MARKET_ID,DEST_CITY_NAME,ORIGIN_CITY_NAME'

        #' | head 10000 ' + \
    body=urllib.parse.urlencode({'search': searchQuery,'output_mode':'csv'})

    #r = requests.post(url, auth=(userName, password),verify = False,data=body)
    #payload = {'search': 'value1', 'key2': 'value2'}
    #params=payload
    r2 = requests.get(url, auth=(userName, password),verify = False,params=body)
    df = pd.read_csv(io.BytesIO((r2.content)),sep=',',header=0)
    return df
    #dd_temp = dd.from_pandas(df,chunksize=10000)
    #results_dd.append(dd_temp)
    #return dd_temp
    
df = dd.from_delayed([get_results(20,i) for i in range(0,20)])
print(df.describe())
print(df.npartitions)
pprint(df.size.compute())