<jupyter_start><jupyter_text>For now 12/8/2020 I recommend using a DataProc cluster to test this notebook, as AI Notebooks Data Hub latest beta has a blocker You'll need to create a service account that has read write rights to BigQuery & Google Cloud Storage, it will also need Admin rights for DataProc.  When you create the cluster use the Preview of the 2.0 version of DataProc cluster which is needed for Koalas<jupyter_code>import databricks.koalas as ks
from pyspark.sql import SparkSession

import datetime<jupyter_output><empty_output><jupyter_text>create the Spark session to distribute processing across a cluster<jupyter_code>spark = SparkSession.builder \
                    .appName('Jupyter BigQuery Storage')\
                    .config('spark.jars', 'gs://spark-lib/bigquery/spark-bigquery-latest_2.12.jar') \
                    .getOrCreate()<jupyter_output><empty_output><jupyter_text>Cluster update function will allow us to add & remove nodes to/from the cluster<jupyter_code>def data_proc_update_cluster(project_id, region, cluster_name, service_account_json, non_preemptibile, preemptibile):
    from google.cloud import dataproc_v1 as dataproc
    from google.oauth2 import service_account

    api_endpoint = {"api_endpoint": f"{region}-dataproc.googleapis.com:443"}
    
    
    credentials = service_account.Credentials.from_service_account_file(service_account_json)
    
    client = dataproc.ClusterControllerClient(credentials=credentials,
                                              client_options=api_endpoint)

    cluster = client.get_cluster(project_id=project_id, 
                                 region=region, 
                                 cluster_name=cluster_name)

    mask = { "paths": {"config.worker_config.num_instances": str(non_preemptibile),
                       "config.secondary_worker_config.num_instances": str(preemptibile),

                      }
           }

    cluster.config.worker_config.num_instances = non_preemptibile # Must update the cluster info itself
    cluster.config.secondary_worker_config.num_instances = preemptibile 

    operation = client.update_cluster(project_id=project_id, 
                                      region=region, 
                                      cluster=cluster,
                                      cluster_name=cluster_name,
                                      update_mask=mask)

    return operation.result()<jupyter_output><empty_output><jupyter_text>Extract data from BigQuery into a Spark dataframe<jupyter_code>table = '`fh-bigquery.reddit_comments.2019_12`'
table = "dustinspark.demos.reddit_comments"

sp_df = spark.read \
             .format("bigquery") \
             .option("table", table) \
             .option('viewsEnabled', True) \
             .load()<jupyter_output><empty_output><jupyter_text>If you want to preview data you can use the toPandas() functionBe sure to limit the number of records before pulling in to Pandas or you can exceed memory<jupyter_code>row_count = sp_df.count()
print(f'{row_count:,} rows in the data frame')
sp_df.limit(10).toPandas()<jupyter_output><empty_output><jupyter_text>Create a Koalas dataframe from the Spark dataframe It's very easy to change between a Spark, Koalas & Pandas dataframe  Be mindful when using Pandas as that will pull all the data in to the master and could exceed memory<jupyter_code>koalas_df = ks.DataFrame(sp_df)
# koala_to_spark_df = koalas_df.to_spark()
# spark_to_pandas_df = sp_df.toPandas()<jupyter_output><empty_output><jupyter_text>Most of the Pandas API is availble with KoalasIn this example you can see how easy it is to work with user defined functions (UDFs)<jupyter_code>def txt_splitter(text):
    if text != None:
        return text.split(' ')
    else:
        return None
    
text = 'Animal Services Office'

txt_splitter(text)

koalas_df['body_2'] = koalas_df['body'].apply(txt_splitter)<jupyter_output><empty_output><jupyter_text>Process some data using 64 CPUs and save it to a GCS bucket<jupyter_code>start = datetime.datetime.now()
koalas_df['body_2'] = koalas_df['body'].str.split(' ')
koalas_df['string_length'] = koalas_df['body'].str.len()
koalas_df['body_lower'] = koalas_df['body'].str.lower()

koala_to_spark_df = koalas_df.to_spark()

koala_to_spark_df.write.format("parquet").option("path", "gs://dusty-study/koalas_run_1").save() # save() is what triggers the DAG to execute

end = datetime.datetime.now()
run_time = end - start
rt_secs = round(run_time.total_seconds())
rt_mins = round(rt_secs / 60)
print(f'64 Core dataframe ran for {rt_secs} seconds / {rt_mins} minutes')<jupyter_output>Pandas dataframe ran for 9487 seconds / 158 minutes<jupyter_text>Let's scale up the cluster 980 CPUs to increase performancePreemptibile instances are 40% of the cost of a non-preemptiable one Costshttps://cloud.google.com/products/calculatorid=0ad8bb87-2d0d-446c-8533-9fee94cb677d1   x Master2   x Workers non-preemptable980 x Workers preemptable1 Hour$79.09 VERSUShttps://cloud.google.com/products/calculatorid=409ff1ba-8e54-4ede-9f5e-f999e19988b81   x Master982 x Workers non-preemptable1 Hour$226.09 SAVINGS **$147** total run cost for 980 premeptable CPUs at \\$1.33  per minute for 20 minutes = \\$26.5<jupyter_code>non_preemptibile = 2
preemptibile = round(980 / 4)
service_account_json = '/home/dustinwilliams/dataproc-jupyter-srvc-acct.json'
username = 'dustinwilliams'
project_id = 'dusty-study'
region = 'us-central1'
cluster_name = 'koalas'
total_cpus = (non_preemptibile + preemptibile) * 4

start = datetime.datetime.now()

update_result = data_proc_update_cluster(project_id, region, cluster_name, service_account_json, non_preemptibile, preemptibile)
%%time
koalas_df['word_array']    = koalas_df['body'].str.split(' ')
koalas_df['string_length'] = koalas_df['body'].str.len()
koalas_df['body_lower']    = koalas_df['body'].str.lower()

koala_to_spark_df = koalas_df.to_spark()

koala_to_spark_df.write.format("parquet").option("path", "gs://dusty-study/koalas_run_2").save() 
non_preemptibile = 2
preemptibile = 0

update_result = data_proc_update_cluster(project_id, region, cluster_name, service_account_json, non_preemptibile, preemptibile)


end = datetime.datetime.now()
run_time = end - start
rt_secs = round(run_time.total_seconds())
rt_mins = round(rt_secs / 60)
print(f'DataProc cluster scaled up to {total_cpus} CPUs and ran for {rt_secs} seconds / {rt_mins} minutes')<jupyter_output>DataProc cluster scaled up to 988 CPUs and ran for 1210 seconds / 20 minutes<jupyter_text>Bigquery can be used to offload processingon demand pricing 1TB = $5Slot reservations can reduce this even more<jupyter_code>start = datetime.datetime.now()
%%bigquery 

CREATE OR REPLACE TABLE `dusty-study.temp.koalas_demo`
AS 
SELECT  body, 
        SPLIT(body, ' ')        AS word_array,
        CHARACTER_LENGTH(body)  AS string_length,
        LOWER(body)             AS body_lower 
FROM    dustinspark.demos.reddit_comments
end = datetime.datetime.now()
run_time = end - start
rt_secs = round(run_time.total_seconds())
rt_mins = round(rt_secs / 60)
print(f'Bigquery query ran for {rt_secs} seconds / {rt_mins} minutes')
%%bigquery

CREATE OR REPLACE MODEL `temp.bqml_example`
OPTIONS(model_type='logistic_reg') AS
SELECT  IF(totals.transactions IS NULL, 0, 1) AS label,
        IFNULL(device.operatingSystem, "") AS os,
        device.isMobile AS is_mobile,
        IFNULL(geoNetwork.country, "") AS country,
        IFNULL(totals.pageviews, 0) AS pageviews
FROM   `bigquery-public-data.google_analytics_sample.ga_sessions_*`
%%bigquery bqml_df_eval

SELECT  *
FROM    ML.EVALUATE(MODEL `temp.bqml_example`, 
          (SELECT  IF(totals.transactions IS NULL, 0, 1)  AS label,
                   IFNULL(device.operatingSystem, "")     AS os,
                   device.isMobile                        AS is_mobile,
                   IFNULL(geoNetwork.country, "")         AS country,
                   IFNULL(totals.pageviews, 0)            AS pageviews
           FROM   `bigquery-public-data.google_analytics_sample.ga_sessions_*`
           LIMIT 10000))
bqml_df_eval.head()
%%bigquery bqml_predictions

WITH ga_data AS (
SELECT  IFNULL(device.operatingSystem, "")  AS os,
        device.isMobile                     AS is_mobile,
        IFNULL(totals.pageviews, 0)         AS pageviews,
        IFNULL(geoNetwork.country, "")      AS country
FROM    `bigquery-public-data.google_analytics_sample.ga_sessions_*`)

SELECT  country,
        SUM(predicted_label) AS total_predicted_purchases
FROM    ML.PREDICT(MODEL `temp.bqml_example`, TABLE ga_data)
GROUP BY country
ORDER BY total_predicted_purchases DESC

bqml_predictions.head()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.09383624655013799, 'fraction_numerical': 0.01782428702851886, 'mean_word_length': 3.7266304347826087, 'pattern_counts': {'":': 4, '<': 39, '<?xml version=': 0, '>': 39, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '10', 'dir': 'jupyter-structured-clean-dedup', 'id': '551641', 'license': 'Apache-2.0', 'n_tokens_mistral': 2971, 'n_tokens_neox': 2784, 'n_words': 647, 'path': 'data_science_notebooks/koalas_BQ_offloading_dataproc_scalingdemo.ipynb', 'repo_name': 'go-dustin/gcp_workflows', 'size': '8696'}