Distributed Data Systems with Azure Databricksやーる

# はじめに
Distributed Data Systems with Azure Databricksやっていきます

https://github.com/PacktPublishing/Distributed-Data-Systems-with-Azure-Databricks

# 開発環境
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/63863/439608bf-be81-a8b2-cbaa-fa4f276a7717.png)

# Chapter06.ipynb

~~~
%fs 
ls /databricks-datasets/structured-streaming/events/
~~~

~~~
path,name,size,modificationTime
dbfs:/databricks-datasets/structured-streaming/events/file-0.json,file-0.json,72530,1469673865000
dbfs:/databricks-datasets/structured-streaming/events/file-1.json,file-1.json,72961,1469673866000
dbfs:/databricks-datasets/structured-streaming/events/file-10.json,file-10.json,73025,1469673878000
...
~~~

~~~
%fs head /databricks-datasets/structured-streaming/events/file-0.json
~~~

~~~
[Truncated to first 65536 bytes]
{"time":1469501107,"action":"Open"}
{"time":1469501147,"action":"Open"}
{"time":1469501202,"action":"Open"}
{"time":1469501219,"action":"Open"}
{"time":1469501225,"action":"Open"}
{"time":1469501234,"action":"Open"}
{"time":1469501245,"action":"Open"}
{"time":1469501246,"action":"Open"}
{"time":1469501248,"action":"Open"}
{"time":1469501256,"action":"Open"}
...
~~~

~~~python
from pyspark.sql.types import *
input_path = "/databricks-datasets/structured-streaming/events/"
~~~

~~~python
json_schema = StructType([StructField("time", TimestampType(), True),
                          StructField("action", StringType(), True)])
~~~

~~~python
static_df = (
  spark
  .read
  .schema(json_schema)
  .json(input_path)
)
~~~

~~~python
display(static_df)
~~~

~~~
time,action
2016-07-28T04:19:28.000+0000,Close
2016-07-28T04:19:28.000+0000,Close
2016-07-28T04:19:29.000+0000,Open
2016-07-28T04:19:31.000+0000,Close
2016-07-28T04:19:31.000+0000,Open
2016-07-28T04:19:31.000+0000,Open
2016-07-28T04:19:32.000+0000,Close
2016-07-28T04:19:33.000+0000,Close
2016-07-28T04:19:35.000+0000,Close
...
~~~

~~~python
from pyspark.sql.functions import *      

static_count_df = (
  static_df
  .groupBy(
    static_df.action,
    window(static_df.time, "1 hour"))
  .count()
)

static_count_df.cache()
~~~

~~~python
static_count_df.createOrReplaceTempView("data_counts")
~~~

~~~
%sql
select action, sum(count) as total_count from data_counts
group by action
~~~

~~~
action,total_count
Open,50000
Close,50000
~~~

![newplot.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/63863/0c3640f5-8998-e83e-d21e-604a5c2a52ee.png)

~~~
%sql
select action, date_format(window.end, "MMM-dd HH:mm") as time, count from data_counts order by time, action
~~~

~~~
action,time,count
Close,Jul-26 03:00,11
Open,Jul-26 03:00,179
Close,Jul-26 04:00,344
Open,Jul-26 04:00,1001
Close,Jul-26 05:00,815
Open,Jul-26 05:00,999
Close,Jul-26 06:00,1003
Open,Jul-26 06:00,1000
Close,Jul-26 07:00,1011
...
~~~

![newplot (1).png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/63863/928d55c0-9743-579a-c966-988afe707ca6.png)

~~~python
from pyspark.sql.functions import *
streaming_df = (
  spark
  .readStream
  .schema(json_schema)               
  .option("maxFilesPerTrigger", 1)  
  .json(input_path)
)

spark.conf.set("spark.sql.shuffle.partitions", "2")  
~~~

~~~python
streaming_counts_df = (                 
  streaming_df
  .groupBy(
    streaming_df.action,
    window(streaming_df.time, "1 hour"))
  .count()
)
~~~

~~~python
streaming_df.isStreaming
~~~

~~~
True
~~~

~~~python
spark.conf.set("spark.sql.shuffle.partitions", "2")  
~~~

~~~python
query = streaming_counts_df.writeStream.format("memory").queryName("data_counts").outputMode("complete").start()
~~~

# Chapter07.ipynb

~~~python
from pyspark.sql.types import *
~~~

~~~python
data = [(1, 'c1', 'a1'), (2, 'c2', 'a2')]
columns = ['id', 'val1', 'val2']
df = spark.createDataFrame(data, columns)
~~~

~~~python
df.printSchema()
df.show(truncate=False)
~~~

~~~
root
 |-- id: long (nullable = true)
 |-- val1: string (nullable = true)
 |-- val2: string (nullable = true)

+---+----+----+
|id |val1|val2|
+---+----+----+
|1  |c1  |a1  |
|2  |c2  |a2  |
+---+----+----+
~~~

~~~python
data_new = [(3, 'c2', 'a2'), (4, 'c3', 'a3')]
columns_new = ['id', 'val1', 'val2']
df_new =spark.createDataFrame(
    data_new,
    columns_new
)
~~~

~~~python
df_new.printSchema()
df_new.show(truncate=False)
~~~

~~~
root
 |-- id: long (nullable = true)
 |-- val1: string (nullable = true)
 |-- val2: string (nullable = true)

+---+----+----+
|id |val1|val2|
+---+----+----+
|3  |c2  |a2  |
|4  |c3  |a3  |
+---+----+----+
~~~

~~~python
union_df = df.union(df_new)
union_df.show(truncate=False)
~~~

~~~
+---+----+----+
|id |val1|val2|
+---+----+----+
|1  |c1  |a1  |
|2  |c2  |a2  |
|3  |c2  |a2  |
|4  |c3  |a3  |
+---+----+----+
~~~

~~~python
union_df.select("val1").show()
~~~

~~~
+----+
|val1|
+----+
|  c1|
|  c2|
|  c2|
|  c3|
+----+
~~~

~~~python
union_df.select("id", "val1").show()
~~~

~~~
+---+----+
| id|val1|
+---+----+
|  1|  c1|
|  2|  c2|
|  3|  c2|
|  4|  c3|
+---+----+
~~~

~~~python
nested_df = union_df
~~~

~~~python
union_df.filter(union_df.val1 == "c2").show(truncate=False)
~~~

~~~
+---+----+----+
|id |val1|val2|
+---+----+----+
|2  |c2  |a2  |
|3  |c2  |a2  |
+---+----+----+
~~~

~~~python
from pyspark.sql.functions import col, asc
filter_df = union_df.filter((col("id") == "3") | (col("val1") == "c2")).sort(asc("id"))
display(filter_df)
~~~

~~~
id,val1,val2
3,c2,a2
~~~

~~~python
dbutils.fs.rm("/tmp/filter_results.parquet", True)
filter_df.write.parquet("/tmp/filter_results.parquet")
~~~

~~~python
dbutils.fs.rm("/tmp/filter_results.parquet", True)
filter_df.write.partitionBy("id", "val1").parquet("/tmp/filter_results.parquet")
~~~

~~~python
non_nun_union = union_df.fillna("-")
~~~

~~~python
non_nun_union = non_nun_union.withColumn("id",col("id").cast("Integer"))
~~~

~~~python
from pyspark.sql.functions import countDistinct
count_distinct_df = non_nun_union.select("id", "val1").groupBy("id").agg(countDistinct ("val1").alias("distinct_val1"))
display(count_distinct_df)
~~~

~~~
id,distinct_val1
1,1
3,1
4,1
2,1
~~~

~~~python
count_distinct_df.describe().show()
~~~

~~~
+-------+------------------+-------------+
|summary|                id|distinct_val1|
+-------+------------------+-------------+
|  count|                 4|            4|
|   mean|               2.5|          1.0|
| stddev|1.2909944487358056|          0.0|
|    min|                 1|            1|
|    max|                 4|            1|
+-------+------------------+-------------+
~~~

~~~python
count_distinct_df.createOrReplaceTempView("count_distinct")
~~~

~~~python
select_sql = spark.sql('''
  SELECT * FROM count_distinct
''')
~~~

~~~python
count_distinct_df = table("count_distinct")
~~~

~~~python
count_distinct_df.cache()
~~~

~~~
Out[28]: DataFrame[id: int, distinct_val1: bigint]
~~~

~~~python
sqlContext.clearCache()
sqlContext.cacheTable("count_distinct")
sqlContext.uncacheTable("count_distinct")
~~~

~~~

~~~


~~~python
import numpy as np
# import pandas as pd
import pyspark.pandas as ps
import databricks.koalas as ks
~~~

~~~python
pandas_series = pd.Series([1, 10, 5, np.nan, 10, 2]) 
~~~

~~~python
koala_series = ks.Series([1, 10, 5, np.nan, 10, 2])
~~~

~~~python
koala_series = ks.Series(pandas_series)
~~~

~~~python
koala_series = ks.from_pandas(pandas_series)
~~~

~~~python
koala_series.sort_index() 
~~~

~~~
Out[36]: 0     1.0
1    10.0
2     5.0
3     NaN
4    10.0
5     2.0
dtype: float64
~~~

~~~python
pandas_df = pd.DataFrame({'val1': np.random.rand(5), 'val2': np.random.rand(5)})
~~~

~~~python
koalas_df = ks.DataFrame({'val1': np.random.rand(5), 'val2': np.random.rand(5)})
~~~

~~~python
koalas_df = ks.DataFrame(pandas_df)
~~~

~~~python
koalas_df = ks.from_pandas(pandas_df)
~~~

~~~python
koalas_df.head(10)
~~~

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/63863/b7350ebf-49a9-8124-bae5-9af1dab7abba.png)

~~~python
koalas_df.describe()
~~~

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/63863/99b7b4d4-174a-48f3-0174-62f5ec1e04ff.png)

~~~python
koalas_df.transpose()
~~~

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/63863/665ececc-e537-5317-13ee-eee83ce19546.png)

~~~python
from databricks.koalas.config import set_option, get_option
ks.get_option('compute.max_rows')
ks.set_option('compute.max_rows', 2000)
~~~

~~~python
koalas_df['val1']  
~~~

~~~
Out[45]: 2    0.302022
3    0.523358
0    0.379593
1    0.575740
4    0.724049
Name: val1, dtype: float64
~~~

~~~python
koalas_df.val1
~~~

~~~
Out[46]: 2    0.302022
3    0.523358
0    0.379593
1    0.575740
4    0.724049
Name: val1, dtype: float64
~~~

~~~python
koalas_df[['val1', 'val2']]
~~~

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/63863/6b100c5c-f0e1-a78c-5beb-ab4c949654ad.png)

~~~python
koalas_df.loc[1:2]
~~~

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/63863/230b5f24-501d-f0f9-0d5e-9d4f1c196a09.png)

~~~python
koalas_df.iloc[:3, 1:2]
~~~

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/63863/7556917f-2dd0-a48e-8c49-ea38b1b9c8f1.png)

~~~python
ks.set_option('compute.ops_on_diff_frames', True)
koalas_series = ks.Series([200, 350, 150, 400, 250], index=[0, 1, 2, 3, 4])
koalas_df['val3'] = koalas_series
~~~

~~~python
koalas_df.head(10)
~~~

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/63863/d8aaf378-5302-3396-8747-61c12e5cd71d.png)

~~~python
ks.reset_option("compute.ops_on_diff_frames")
~~~

~~~python
koalas_df.apply(np.cumsum)
~~~

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/63863/751626dd-fb55-9607-76ec-6d90a0843c95.png)

~~~python
koalas_df.apply(np.cumsum, axis=1)
~~~

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/63863/54eaa29b-5093-ccc1-f2e0-7d9bca82c12c.png)

~~~python
def square_root(x) -> ks.Series[np.float64]:
  return x ** .5
koalas_df.apply(square_root)
~~~

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/63863/88fa4b68-6450-af87-4284-045653a62dfe.png)

~~~python
ks.set_option('compute.shortcut_limit', 5000)
~~~

~~~python
koalas_df = ks.DataFrame({'val1': [15, 18, 489, 675, 1776], 'val2': [4, 25, 181, 600, 1900]}, index=[1, 2, 3, 4, 5])
koalas_df.plot.line()
~~~

![newplot (2).png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/63863/a137b417-5d59-1391-e6fb-b390b3f91e92.png)

~~~python
koalas_df = pd.DataFrame(np.random.randint(1, 10, 1000), columns=['c1'])
koalas_df ['c2'] = koalas_df ['c1'] + np.random.randint(1, 10, 1000)
koalas_df = ks.from_pandas(koalas_df)
koalas_df.plot.hist(bins=14, alpha=0.5)
~~~

![newplot (3).png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/63863/31671979-6a88-2bf2-d4b3-fa2cbdd31a16.png)

~~~python
koalas_df = ks.DataFrame({'year': [2010, 2011, 2012, 2013, 2014],
                          'c1': [15, 24, 689, 575, 1376],
                          'c2': [4, 27, 311, 720, 1650]})
ks.sql("SELECT * FROM {koalas_df} WHERE c1 > 100")
~~~

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/63863/303618b8-b8bb-26b7-1cc6-69d4af24af85.png)

~~~python
koalas_df = ks.DataFrame({'c1': [1, 2, 3, 4, 5], 'c2': [10, 20, 30, 40, 50]})
spark_df = koalas_df.to_spark()
type(spark_df)
spark_df.show()
~~~

~~~
+---+---+
| c1| c2|
+---+---+
|  1| 10|
|  2| 20|
|  3| 30|
|  4| 40|
|  5| 50|
+---+---+
~~~

~~~python
from databricks.koalas import option_context
with option_context("compute.default_index_type", "distributed-sequence"):
  koalas_df = spark_df.to_koalas()
~~~

~~~python
spark_df.to_koalas(index_col='c1')
~~~

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/63863/5877906f-d177-da35-10cc-5a2d4fbed03e.png)

~~~python
koalas_df.cache()
~~~

~~~
/databricks/python/lib/python3.8/site-packages/databricks/koalas/frame.py:4600: FutureWarning:

DataFrame.cache is deprecated as of DataFrame.spark.cache. Please use the API instead.
~~~

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/63863/9b5708ce-aa1a-30af-2773-b2fb3846ddf5.png)

~~~python
%pip install bokeh
~~~

~~~python
from bokeh.plotting import figure
from bokeh.embed import components, file_html
from bokeh.resources import CDN
~~~

~~~python
x = [1, 2, 3, 4, 5, 6, 7, 8, 9]
y = [2, 4, 3, 4, 5, 6, 7, 9, 8]
~~~

~~~python
p = figure(title="this simple line example", x_axis_label='x_val', y_axis_label='y_val')
~~~

~~~python
p.line(x, y, legend_label="this line.", line_width=2)
~~~

~~~
Out[4]: 
GlyphRenderer(id = '1039', ‹‹‹
              coordinates = None,
              data_source = ColumnDataSource(id='1035', ...),
              glyph = Line(id='1036', ...),
              group = None,
              hover_glyph = None,
              js_event_callbacks = {},
              js_property_callbacks = {},
              level = 'glyph',
              muted = False,
              muted_glyph = Line(id='1038', ...),
              name = None,
              nonselection_glyph = Line(id='1037', ...),
              selection_glyph = 'auto',
              subscribed_events = [],
              syncable = True,
              tags = [],
              view = CDSView(id='1040', ...),
              visible = True,
              x_range_name = 'default',
              y_range_name = 'default')
~~~

~~~python
html = file_html(p, CDN, "my_plot")
~~~

~~~python
displayHTML(html)
~~~

![bokeh_plot.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/63863/1e6d0051-7304-48c6-6a6e-1e288eb473cb.png)

~~~python
import numpy as np
import matplotlib.pyplot as plt
x = np.linspace(0, 2*np.pi, 50)
y = np.sin(x)

fig, ax = plt.subplots()
ax.plot(x, y, 'k--')

ax.set_xlim((0, 2*np.pi))
ax.set_xticks([0, np.pi, 2*np.pi])
ax.set_xticklabels(['0', '$\pi$','2$\pi$'])
ax.set_ylim((-1.5, 1.5))
ax.set_yticks([-1, 0, 1])

display(fig)
~~~

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/63863/a3166e93-4151-1936-608c-60f7ecc4bc04.png)

~~~python
from plotly.offline import plot
from plotly.graph_objs import *
import numpy as np

x = np.random.randn(1000)
y = np.random.randn(1000)

p = plot(
  [Histogram2dContour(x=x, y=y, contours=Contours(coloring='heatmap')),
   Scatter(x=x, y=y, mode='markers', marker=Marker(color='white', size=3, opacity=0.3))
  ],
  output_type='div'
)

displayHTML(p)
~~~

~~~
/databricks/python/lib/python3.8/site-packages/plotly/graph_objs/_deprecations.py:204: DeprecationWarning:

plotly.graph_objs.Contours is deprecated.
Please replace it with one of the following more specific types
  - plotly.graph_objs.contour.Contours
  - plotly.graph_objs.surface.Contours
  - etc.


/databricks/python/lib/python3.8/site-packages/plotly/graph_objs/_deprecations.py:434: DeprecationWarning:

plotly.graph_objs.Marker is deprecated.
Please replace it with one of the following more specific types
  - plotly.graph_objs.scatter.Marker
  - plotly.graph_objs.histogram.selected.Marker
  - etc.
~~~

![newplot (4).png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/63863/8a8d8154-a8c9-cba9-6416-e8e565604f6f.png)

# Chapter08.ipynb

# Chapter09.ipynb

# Chapter10.ipynb

# Chapter11.ipynb

# Chapter12.ipynb