<commit_before>#+TITLE: Apache Spark

* Table of Contents :TOC_3_gh:
- [[#overview][Overview]]
- [[#reference][Reference]]
  - [[#spark-api][Spark API]]
    - [[#file-input-pattern][File Input pattern]]
  - [[#spark-sql][Spark SQL]]
    - [[#reference-links][Reference Links]]
    - [[#examples][Examples]]
    - [[#recipes][Recipes]]
    - [[#example-codes][Example codes]]
- [[#terminology][Terminology]]
- [[#topics][Topics]]
  - [[#how-merging-schema-across-files-works][How merging schema across files works]]
- [[#how-to][How-to]]
- [[#links][Links]]
- [[#staging][Staging]]
  - [[#2017-08-21-mon-required-iam-role-for-using-s3-with-spark][<2017-08-21 Mon> Required IAM Role for using S3 with Spark]]
  - [[#2017-07-21-fri-hiveql-window-function][<2017-07-21 Fri> HiveQL, Window Function]]
  - [[#2017-07-21-fri-retention-query][<2017-07-21 Fri> Retention Query]]

* Overview
* Reference
** Spark API
*** File Input pattern
http://stackoverflow.com/questions/31782763/how-to-use-regex-to-include-exclude-some-input-files-in-sc-textfile

- ~*~ (match 0 or more character)
- ~?~ (match single character)
- ~[ab]~ (character class)
- ~[^ab]~ (negated character class)
- ~[a-b]~ (character range)
- ~{a,b}~ (alternation)
- ~\c~ (escape character)

You can use commas as delimiters of multiple patterns:
: sc.textFile("/user/Orders/2015072[7-9]*,/user/Orders/2015073[0-1]*")

Which is same as:
: sc.textFile("/user/Orders/201507{2[7-9],3[0-1]}*")

** Spark SQL
*** Reference Links
There seems to be a newly documented page:
https://docs.databricks.com/spark/latest/spark-sql/index.html

Spark SQL's query languages is based on ~HiveQL~.
- https://cwiki.apache.org/confluence/display/Hive/LanguageManual+Select
- https://cwiki.apache.org/confluence/display/Hive/LanguageManual+UDF
- http://spark.apache.org/docs/latest/sql-programming-guide.html#supported-hive-features

*** Examples
#+BEGIN_SRC sql
  -- basics
  select col1 from t1 where col1 > 10
  select * from t1 limit 5

  -- plan
  explain select * from t1

  -- group by
  select col1, count(*) from t1 group by col1
  select col1, sum(col2) from t1 group by col1

  -- 'group by' can be specified with position numbers(1-indexed from selected columns)
  select col1, sum(col2) from t1 group by 1

  -- distinct
  select col1, col2 from t1
  1 3
  1 3
  1 4
  2 5

  select distinct col1, col2 from t1
  1 3
  1 4
  2 5

  select distinct col1 from t1
  1
  2

  -- distinct can be used within 'count'
  select col1, count(distinct col2) from t1 group by col1

  -- having
  select col1 from t1 group by col1 having sum(col2) > 10
  -- same as above
  select col1 from (select col1, sum(col2) as col2sum from t1 group by col1) t2 where t2.col2sum > 10

  -- order by
  select col1 from t1 order by col1 desc

  -- join
  select a.* from a join b on (a.id = b.id and a.department = b.department)
  select a.* from a left outer join b on (a.id <> b.id)
  select a.val, b.val, c.val from a join b on (a.key = b.key1) join c on (c.key = b.key1)

  -- joins occur before where clauses, this is a bad way:
  select a.val, b.val from a left outer join b on (a.key=b.key)
  where a.ds='2009-07-07' and b.ds='2009-07-07'

  -- following is better:
  select a.val, b.val from a left outer join b
  on (a.key=b.key and b.ds='2009-07-07' and a.ds='2009-07-07')

  -- union
  select u.id, actions.date
  from (
    select av.uid as uid
    from action_video av
    where av.date = '2008-06-03'
    union all
    select ac.uid as uid
    from action_comment ac
    where ac.date = '2008-06-03'
  ) actions join users u on (u.id = actions.uid)

  -- if, case
  select if(field in (0, 1), 'ab', 'c') from tbl

  select
    case field
    when 0 then 'a'
    when 1 then 'b'
    else 'c'
    end
  from tbl

  -- subqueries
  select col
  from (
    select a+b as col
    from t1
  ) t2

  select *
  from a
  where a.a in (select foo from b);

  select a
  from t1
  where exists (select b from t2 where t1.x = t2.y)

  -- common table expression
  with q1 as (select key from src where key = '5')
  select *
  from q1;

  with q1 as (select * from src where key= '5'),
       q2 as (select * from src s2 where key = '4')
  select * from q1 union all select * from q2;

  -- create table as select example
  create table s2 as
  with q1 as ( select key from src where key = '4')
  select * from q1;

  -- create or replace temporary view is recommended instead of just 'create table'
  create or replace temporary view foo as select * from t1 limit 1

  -- view example
  create view v1 as
  with q1 as ( select key from src where key = '5')
  select * from q1;

  -- lateral view
  select adid, count(1)
  from pageads lateral view explode(adid_list) adtable as adid
  group by adid

  select k, v
  from tbl lateral view explode(kvmap) kvs as k, v
  group by k

  select mycol1, mycol2 from basetable
  lateral view explode(col1) mytable1 as mycol1
  lateral view explode(col2) mytable2 as mycol2;

  select * from src lateral view outer explode(array()) c as a limit 10;

  -- time range (t is of timestamp type)
  select t from table1
  where t > to_utc_timestamp("2016-12-25", "UTC")
  and t < to_utc_timestamp("2016-12-25 12:00", "UTC")

  -- timestamp to string
  select date_format(t, 'YYYY-MM-dd') from tbl

  -- select field with special characters(use backtick)
  select `@time` from t1

  -- concat_ws to make an array as a string
  -- map_values to make a map as an array
  -- <array of structtype>.<field> goes into an <array of field>
  select concat_ws(", ", map_values(items).price)
  from Items
#+END_SRC

*** Recipes
**** Referencing query results as ~DataFrame~ in spark application
#+BEGIN_SRC sql
  %sql
  create or replace temporary view foo as select * from t1 limit 1
#+END_SRC
#+BEGIN_SRC scala
  val spark: SparkSession = ...
  val df = spark.table("foo")
  // work with df
#+END_SRC

*** Example codes
https://github.com/apache/spark/tree/master/examples/src/main/scala/org/apache/spark/examples/sql

* Terminology
* Topics
** How merging schema across files works
- https://spark.apache.org/docs/latest/sql-programming-guide.html#schema-merging

There were no clear documentation about how the merging schema across files works.
There are only some the general guidelines of growing schema: Append only, no modification.

But I was curious that if appending is ok, can I just reorder the ~StructFields~?
Because ~StructType~ takes them as a ~List~, it may. But as the data source can be ~json~,
which doesn't care the order of fields, it may not. So I tested.

The conclusion is following:
- The order of ~StructField~ doesn't matter. *Only field name matters.*
- If ~mergeSchema~ is ~true~, all fields are merged.
- If ~mergeSchema~ is ~false~, which is default, The schema of the first file in alphabetical order has priority 
- The ~nullable~ in ~StructType~ doesn't matter. It seemm to only matter when processing raw data, but merging files.
- *If there are fields with the same name and different types, schema merging will cause runtime erros.*

#+BEGIN_SRC scala
  import org.apache.spark.sql.Row
  import org.apache.spark.sql.types._
  import spark.implicits._

  def createSchema(schema: String): StructType = {
      val fieldNames = schema.split(" ")
      val fields = fieldNames.map { name =>
          StructField(name, StringType)
      }
      StructType(fields)
  }

  // A schema of 'name' and 'addr'
  val schema1 = createSchema("name addr")
  val data1   = List(Row("yeonghoey", "jamsil"))
  val rdd1    = spark.sparkContext.parallelize(data1)
  val df1     = spark.createDataFrame(rdd1, schema1)
  df1.write.mode("overwrite").parquet("data1")

  // Add 'sex' field in between the fields of schema1.
  val schema2 = createSchema("name sex addr")
  val data2   = List(Row("cwkim", "male", "unjung"))
  val rdd2    = spark.sparkContext.parallelize(data2)
  val df2     = spark.createDataFrame(rdd2, schema2)
  df2.write.mode("overwrite").parquet("data2")

  // Append 'sex' field to the schema1.
  val schema3 = createSchema("name addr sex")
  val data3   = List(Row("sub", "yangjae", "male"))
  val rdd3    = spark.sparkContext.parallelize(data3)
  val df3     = spark.createDataFrame(rdd3, schema3)
  df3.write.mode("overwrite").parquet("data3")

  // A schema of 'name' and 'sex'
  val schema4 = createSchema("name sex")
  val data4   = List(Row("suminb", "male"))
  val rdd4    = spark.sparkContext.parallelize(data4)
  val df4     = spark.createDataFrame(rdd4, schema4)
  df4.write.mode("overwrite").parquet("data4")

  // ----------------------------------------------------------------------

  spark.read.parquet("data{1,2}").show()
  // +---------+------+
  // |     name|  addr|
  // +---------+------+
  // |    cwkim|unjung|
  // |yeonghoey|jamsil|
  // +---------+------+

  spark.read.option("mergeSchema", true).parquet("data{1,2}").show()
  // +---------+------+----+
  // |     name|  addr| sex|
  // +---------+------+----+
  // |    cwkim|unjung|male|
  // |yeonghoey|jamsil|null|
  // +---------+------+----+

  spark.read.parquet("data{1,3}").show()
  // +---------+-------+
  // |     name|   addr|
  // +---------+-------+
  // |      sub|yangjae|
  // |yeonghoey| jamsil|
  // +---------+-------+

  spark.read.option("mergeSchema", true).parquet("data{1,3}").show()
  // +---------+-------+----+
  // |     name|   addr| sex|
  // +---------+-------+----+
  // |      sub|yangjae|male|
  // |yeonghoey| jamsil|null|
  // +---------+-------+----+

  spark.read.parquet("data{2,3}").show()
  // +-----+----+-------+
  // | name| sex|   addr|
  // +-----+----+-------+
  // |cwkim|male| unjung|
  // |  sub|male|yangjae|
  // +-----+----+-------+

  spark.read.option("mergeSchema", true).parquet("data{2,3}").show()
  // +-----+----+-------+
  // | name| sex|   addr|
  // +-----+----+-------+
  // |cwkim|male| unjung|
  // |  sub|male|yangjae|
  // +-----+----+-------+

  spark.read.parquet("data{3,4}").show()
  // +------+-------+----+
  // |  name|   addr| sex|
  // +------+-------+----+
  // |   sub|yangjae|male|
  // |suminb|   null|male|
  // +------+-------+----+

  spark.read.option("mergeSchema", true).parquet("data{3,4}").show()
  // +------+-------+----+
  // |  name|   addr| sex|
  // +------+-------+----+
  // |   sub|yangjae|male|
  // |suminb|   null|male|
  // +------+-------+----+

  spark.read.parquet("data{1,4}").show()
  // +---------+------+
  // |     name|  addr|
  // +---------+------+
  // |yeonghoey|jamsil|
  // |   suminb|  null|
  // +---------+------+

  spark.read.option("mergeSchema", true).parquet("data{1,4}").show()
  // +---------+------+----+
  // |     name|  addr| sex|
  // +---------+------+----+
  // |yeonghoey|jamsil|null|
  // |   suminb|  null|male|
  // +---------+------+----+

  spark.read.parquet("data{1,2,3,4}").show()
  // +---------+-------+
  // |     name|   addr|
  // +---------+-------+
  // |    cwkim| unjung|
  // |      sub|yangjae|
  // |yeonghoey| jamsil|
  // |   suminb|   null|
  // +---------+-------+

  spark.read.option("mergeSchema", true).parquet("data{1,2,3,4}").show()
  // +---------+-------+----+
  // |     name|   addr| sex|
  // +---------+-------+----+
  // |    cwkim| unjung|male|
  // |      sub|yangjae|male|
  // |yeonghoey| jamsil|null|
  // |   suminb|   null|male|
  // +---------+-------+----+
#+END_SRC

#+BEGIN_SRC scala
  import org.apache.spark.sql.Row
  import org.apache.spark.sql.types._
  import spark.implicits._

  val schema1 = StructType(List(StructField("x'", StringType)))
  val data1   = List(Row("yeonghoey"))
  val rdd1    = spark.sparkContext.parallelize(data1)
  val df1     = spark.createDataFrame(rdd1, schema1)
  df1.write.mode("overwrite").parquet("data1")

  val schema2 = StructType(List(StructField("x'", IntegerType)))
  val data2   = List(Row(31))
  val rdd2    = spark.sparkContext.parallelize(data2)
  val df2     = spark.createDataFrame(rdd2, schema2)
  df2.write.mode("overwrite").parquet("data2")

  // ----------------------------------------------------------------------
  
  spark.read.parquet("data{1,2}").show()
  // Caused by: java.lang.UnsupportedOperationException: Unimplemented type: StringType

  spark.read.option("mergeSchema", true).parquet("data{1,2}").show()
  // Caused by: org.apache.spark.SparkException: Failed to merge incompatible data types StringType and IntegerType
#+END_SRC

* How-to
* Links
* Staging
** TODO <2017-08-21 Mon> Required IAM Role for using S3 with Spark
- https://docs.databricks.com/user-guide/cloud-configurations/aws/iam-roles.html
  For using spark with data in s3, just s3 Programmatic Access policy required

** TODO <2017-07-21 Fri> HiveQL, Window Function
- https://cwiki.apache.org/confluence/display/Hive/LanguageManual+WindowingAndAnalytics

** TODO <2017-07-21 Fri> Retention Query
- https://blog.treasuredata.com/blog/2016/07/22/rolling-retention-done-right-in-sql/

#+BEGIN_SRC sql
  %sql
  create or replace temporary view
    RetentionByCount
  as with

  daily as (
    select distinct
      user_id,
      to_date(from_utc_timestamp(`@time`, "JST")) as day
    from UserLogins
  ),

  by_first_day as (
    select
      user_id,
      day,
      first_value(day) over (partition by user_id order by day) as first_day
    from daily
  ),

  by_diff as (
    select
      first_day,
      datediff(day, first_day) as diff
    from by_first_day
  )

  select
    first_day,
    sum(case when diff = 0 then 1 else 0 end) as day00,
    sum(case when diff = 1 then 1 else 0 end) as day01,
    sum(case when diff = 2 then 1 else 0 end) as day02,
    sum(case when diff = 3 then 1 else 0 end) as day03,
    sum(case when diff = 4 then 1 else 0 end) as day04,
    sum(case when diff = 5 then 1 else 0 end) as day05,
    sum(case when diff = 6 then 1 else 0 end) as day06,
    sum(case when diff = 7 then 1 else 0 end) as day07,
    sum(case when diff = 8 then 1 else 0 end) as day08,
    sum(case when diff = 9 then 1 else 0 end) as day09,
    sum(case when diff = 10 then 1 else 0 end) as day10,
    sum(case when diff = 11 then 1 else 0 end) as day11,
    sum(case when diff = 12 then 1 else 0 end) as day12,
    sum(case when diff = 13 then 1 else 0 end) as day13,
    sum(case when diff = 14 then 1 else 0 end) as day14,
    sum(case when diff = 15 then 1 else 0 end) as day15,
    sum(case when diff = 16 then 1 else 0 end) as day16,
    sum(case when diff = 17 then 1 else 0 end) as day17,
    sum(case when diff = 18 then 1 else 0 end) as day18,
    sum(case when diff = 19 then 1 else 0 end) as day19,
    sum(case when diff = 20 then 1 else 0 end) as day20,
    sum(case when diff = 21 then 1 else 0 end) as day21,
    sum(case when diff = 22 then 1 else 0 end) as day22,
    sum(case when diff = 23 then 1 else 0 end) as day23,
    sum(case when diff = 24 then 1 else 0 end) as day24,
    sum(case when diff = 25 then 1 else 0 end) as day25,
    sum(case when diff = 26 then 1 else 0 end) as day26,
    sum(case when diff = 27 then 1 else 0 end) as day27,
    sum(case when diff = 28 then 1 else 0 end) as day28,
    sum(case when diff = 29 then 1 else 0 end) as day29,
    sum(case when diff = 30 then 1 else 0 end) as day30
  from by_diff
  group by 1
  order by 1
#+END_SRC

#+BEGIN_SRC sql
  %sql
  create or replace temporary view 
    RetentionByPercentage
  as
  select
    first_day,
    day00 as `new`,
    round(day01 / day00 * 100, 2) as `d+1`,
    round(day02 / day00 * 100, 2) as `d+2`,
    round(day03 / day00 * 100, 2) as `d+3`,
    round(day04 / day00 * 100, 2) as `d+4`,
    round(day05 / day00 * 100, 2) as `d+5`,
    round(day06 / day00 * 100, 2) as `d+6`,
    round(day07 / day00 * 100, 2) as `d+7`,
    round(day08 / day00 * 100, 2) as `d+8`,
    round(day09 / day00 * 100, 2) as `d+9`,
    round(day10 / day00 * 100, 2) as `d+10`,
    round(day11 / day00 * 100, 2) as `d+11`,
    round(day12 / day00 * 100, 2) as `d+12`,
    round(day13 / day00 * 100, 2) as `d+13`,
    round(day14 / day00 * 100, 2) as `d+14`,
    round(day15 / day00 * 100, 2) as `d+15`,
    round(day16 / day00 * 100, 2) as `d+16`,
    round(day17 / day00 * 100, 2) as `d+17`,
    round(day18 / day00 * 100, 2) as `d+18`,
    round(day19 / day00 * 100, 2) as `d+19`,
    round(day20 / day00 * 100, 2) as `d+20`,
    round(day21 / day00 * 100, 2) as `d+21`,
    round(day22 / day00 * 100, 2) as `d+22`,
    round(day23 / day00 * 100, 2) as `d+23`,
    round(day24 / day00 * 100, 2) as `d+24`,
    round(day25 / day00 * 100, 2) as `d+25`,
    round(day26 / day00 * 100, 2) as `d+26`,
    round(day27 / day00 * 100, 2) as `d+27`,
    round(day28 / day00 * 100, 2) as `d+28`,
    round(day29 / day00 * 100, 2) as `d+29`,
    round(day30 / day00 * 100, 2) as `d+30`
  from RetentionByCount
  order by 1
#+END_SRC
<commit_msg>Update spark<commit_after>#+TITLE: Apache Spark

* Table of Contents :TOC_3_gh:
- [[#overview][Overview]]
- [[#reference][Reference]]
  - [[#spark-api][Spark API]]
    - [[#file-input-pattern][File Input pattern]]
  - [[#spark-sql][Spark SQL]]
    - [[#reference-links][Reference Links]]
    - [[#examples][Examples]]
    - [[#recipes][Recipes]]
    - [[#example-codes][Example codes]]
- [[#terminology][Terminology]]
- [[#topics][Topics]]
  - [[#how-schema-merging-works][How schema merging works]]
- [[#how-to][How-to]]
- [[#links][Links]]
- [[#staging][Staging]]
  - [[#2017-08-21-mon-required-iam-role-for-using-s3-with-spark][<2017-08-21 Mon> Required IAM Role for using S3 with Spark]]
  - [[#2017-07-21-fri-hiveql-window-function][<2017-07-21 Fri> HiveQL, Window Function]]
  - [[#2017-07-21-fri-retention-query][<2017-07-21 Fri> Retention Query]]

* Overview
* Reference
** Spark API
*** File Input pattern
http://stackoverflow.com/questions/31782763/how-to-use-regex-to-include-exclude-some-input-files-in-sc-textfile

- ~*~ (match 0 or more character)
- ~?~ (match single character)
- ~[ab]~ (character class)
- ~[^ab]~ (negated character class)
- ~[a-b]~ (character range)
- ~{a,b}~ (alternation)
- ~\c~ (escape character)

You can use commas as delimiters of multiple patterns:
: sc.textFile("/user/Orders/2015072[7-9]*,/user/Orders/2015073[0-1]*")

Which is same as:
: sc.textFile("/user/Orders/201507{2[7-9],3[0-1]}*")

** Spark SQL
*** Reference Links
There seems to be a newly documented page:
https://docs.databricks.com/spark/latest/spark-sql/index.html

Spark SQL's query languages is based on ~HiveQL~.
- https://cwiki.apache.org/confluence/display/Hive/LanguageManual+Select
- https://cwiki.apache.org/confluence/display/Hive/LanguageManual+UDF
- http://spark.apache.org/docs/latest/sql-programming-guide.html#supported-hive-features

*** Examples
#+BEGIN_SRC sql
  -- basics
  select col1 from t1 where col1 > 10
  select * from t1 limit 5

  -- plan
  explain select * from t1

  -- group by
  select col1, count(*) from t1 group by col1
  select col1, sum(col2) from t1 group by col1

  -- 'group by' can be specified with position numbers(1-indexed from selected columns)
  select col1, sum(col2) from t1 group by 1

  -- distinct
  select col1, col2 from t1
  1 3
  1 3
  1 4
  2 5

  select distinct col1, col2 from t1
  1 3
  1 4
  2 5

  select distinct col1 from t1
  1
  2

  -- distinct can be used within 'count'
  select col1, count(distinct col2) from t1 group by col1

  -- having
  select col1 from t1 group by col1 having sum(col2) > 10
  -- same as above
  select col1 from (select col1, sum(col2) as col2sum from t1 group by col1) t2 where t2.col2sum > 10

  -- order by
  select col1 from t1 order by col1 desc

  -- join
  select a.* from a join b on (a.id = b.id and a.department = b.department)
  select a.* from a left outer join b on (a.id <> b.id)
  select a.val, b.val, c.val from a join b on (a.key = b.key1) join c on (c.key = b.key1)

  -- joins occur before where clauses, this is a bad way:
  select a.val, b.val from a left outer join b on (a.key=b.key)
  where a.ds='2009-07-07' and b.ds='2009-07-07'

  -- following is better:
  select a.val, b.val from a left outer join b
  on (a.key=b.key and b.ds='2009-07-07' and a.ds='2009-07-07')

  -- union
  select u.id, actions.date
  from (
    select av.uid as uid
    from action_video av
    where av.date = '2008-06-03'
    union all
    select ac.uid as uid
    from action_comment ac
    where ac.date = '2008-06-03'
  ) actions join users u on (u.id = actions.uid)

  -- if, case
  select if(field in (0, 1), 'ab', 'c') from tbl

  select
    case field
    when 0 then 'a'
    when 1 then 'b'
    else 'c'
    end
  from tbl

  -- subqueries
  select col
  from (
    select a+b as col
    from t1
  ) t2

  select *
  from a
  where a.a in (select foo from b);

  select a
  from t1
  where exists (select b from t2 where t1.x = t2.y)

  -- common table expression
  with q1 as (select key from src where key = '5')
  select *
  from q1;

  with q1 as (select * from src where key= '5'),
       q2 as (select * from src s2 where key = '4')
  select * from q1 union all select * from q2;

  -- create table as select example
  create table s2 as
  with q1 as ( select key from src where key = '4')
  select * from q1;

  -- create or replace temporary view is recommended instead of just 'create table'
  create or replace temporary view foo as select * from t1 limit 1

  -- view example
  create view v1 as
  with q1 as ( select key from src where key = '5')
  select * from q1;

  -- lateral view
  select adid, count(1)
  from pageads lateral view explode(adid_list) adtable as adid
  group by adid

  select k, v
  from tbl lateral view explode(kvmap) kvs as k, v
  group by k

  select mycol1, mycol2 from basetable
  lateral view explode(col1) mytable1 as mycol1
  lateral view explode(col2) mytable2 as mycol2;

  select * from src lateral view outer explode(array()) c as a limit 10;

  -- time range (t is of timestamp type)
  select t from table1
  where t > to_utc_timestamp("2016-12-25", "UTC")
  and t < to_utc_timestamp("2016-12-25 12:00", "UTC")

  -- timestamp to string
  select date_format(t, 'YYYY-MM-dd') from tbl

  -- select field with special characters(use backtick)
  select `@time` from t1

  -- concat_ws to make an array as a string
  -- map_values to make a map as an array
  -- <array of structtype>.<field> goes into an <array of field>
  select concat_ws(", ", map_values(items).price)
  from Items
#+END_SRC

*** Recipes
**** Referencing query results as ~DataFrame~ in spark application
#+BEGIN_SRC sql
  %sql
  create or replace temporary view foo as select * from t1 limit 1
#+END_SRC
#+BEGIN_SRC scala
  val spark: SparkSession = ...
  val df = spark.table("foo")
  // work with df
#+END_SRC

*** Example codes
https://github.com/apache/spark/tree/master/examples/src/main/scala/org/apache/spark/examples/sql

* Terminology
* Topics
** How schema merging works
- https://spark.apache.org/docs/latest/sql-programming-guide.html#schema-merging

There were no clear documentation about how the merging schema across files works.
There are only some the general guidelines of growing schema: Append only, no modification.

But I was curious that if appending is ok, can I just reorder the ~StructFields~?
Because ~StructType~ takes them as a ~List~, it may. But as the data source can be ~json~,
which doesn't care the order of fields, it may not. So I tested.

The conclusion is following:
- The order of ~StructField~ doesn't matter. *Only field name matters.*
- If ~mergeSchema~ is ~true~, all fields are merged.
- If ~mergeSchema~ is ~false~, which is default, The schema of the first file in alphabetical order has priority 
- The ~nullable~ in ~StructType~ doesn't matter. It seemm to only matter when processing raw data, but merging files.
- *If there are fields with the same name and different types, schema merging will cause runtime erros.*

#+BEGIN_SRC scala
  import org.apache.spark.sql.Row
  import org.apache.spark.sql.types._
  import spark.implicits._

  def createSchema(schema: String): StructType = {
      val fieldNames = schema.split(" ")
      val fields = fieldNames.map { name =>
          StructField(name, StringType)
      }
      StructType(fields)
  }

  // A schema of 'name' and 'addr'
  val schema1 = createSchema("name addr")
  val data1   = List(Row("yeonghoey", "jamsil"))
  val rdd1    = spark.sparkContext.parallelize(data1)
  val df1     = spark.createDataFrame(rdd1, schema1)
  df1.write.mode("overwrite").parquet("data1")

  // Add 'sex' field in between the fields of schema1.
  val schema2 = createSchema("name sex addr")
  val data2   = List(Row("cwkim", "male", "unjung"))
  val rdd2    = spark.sparkContext.parallelize(data2)
  val df2     = spark.createDataFrame(rdd2, schema2)
  df2.write.mode("overwrite").parquet("data2")

  // Append 'sex' field to the schema1.
  val schema3 = createSchema("name addr sex")
  val data3   = List(Row("sub", "yangjae", "male"))
  val rdd3    = spark.sparkContext.parallelize(data3)
  val df3     = spark.createDataFrame(rdd3, schema3)
  df3.write.mode("overwrite").parquet("data3")

  // A schema of 'name' and 'sex'
  val schema4 = createSchema("name sex")
  val data4   = List(Row("suminb", "male"))
  val rdd4    = spark.sparkContext.parallelize(data4)
  val df4     = spark.createDataFrame(rdd4, schema4)
  df4.write.mode("overwrite").parquet("data4")

  // ----------------------------------------------------------------------

  spark.read.parquet("data{1,2}").show()
  // +---------+------+
  // |     name|  addr|
  // +---------+------+
  // |    cwkim|unjung|
  // |yeonghoey|jamsil|
  // +---------+------+

  spark.read.option("mergeSchema", true).parquet("data{1,2}").show()
  // +---------+------+----+
  // |     name|  addr| sex|
  // +---------+------+----+
  // |    cwkim|unjung|male|
  // |yeonghoey|jamsil|null|
  // +---------+------+----+

  spark.read.parquet("data{1,3}").show()
  // +---------+-------+
  // |     name|   addr|
  // +---------+-------+
  // |      sub|yangjae|
  // |yeonghoey| jamsil|
  // +---------+-------+

  spark.read.option("mergeSchema", true).parquet("data{1,3}").show()
  // +---------+-------+----+
  // |     name|   addr| sex|
  // +---------+-------+----+
  // |      sub|yangjae|male|
  // |yeonghoey| jamsil|null|
  // +---------+-------+----+

  spark.read.parquet("data{2,3}").show()
  // +-----+----+-------+
  // | name| sex|   addr|
  // +-----+----+-------+
  // |cwkim|male| unjung|
  // |  sub|male|yangjae|
  // +-----+----+-------+

  spark.read.option("mergeSchema", true).parquet("data{2,3}").show()
  // +-----+----+-------+
  // | name| sex|   addr|
  // +-----+----+-------+
  // |cwkim|male| unjung|
  // |  sub|male|yangjae|
  // +-----+----+-------+

  spark.read.parquet("data{3,4}").show()
  // +------+-------+----+
  // |  name|   addr| sex|
  // +------+-------+----+
  // |   sub|yangjae|male|
  // |suminb|   null|male|
  // +------+-------+----+

  spark.read.option("mergeSchema", true).parquet("data{3,4}").show()
  // +------+-------+----+
  // |  name|   addr| sex|
  // +------+-------+----+
  // |   sub|yangjae|male|
  // |suminb|   null|male|
  // +------+-------+----+

  spark.read.parquet("data{1,4}").show()
  // +---------+------+
  // |     name|  addr|
  // +---------+------+
  // |yeonghoey|jamsil|
  // |   suminb|  null|
  // +---------+------+

  spark.read.option("mergeSchema", true).parquet("data{1,4}").show()
  // +---------+------+----+
  // |     name|  addr| sex|
  // +---------+------+----+
  // |yeonghoey|jamsil|null|
  // |   suminb|  null|male|
  // +---------+------+----+

  spark.read.parquet("data{1,2,3,4}").show()
  // +---------+-------+
  // |     name|   addr|
  // +---------+-------+
  // |    cwkim| unjung|
  // |      sub|yangjae|
  // |yeonghoey| jamsil|
  // |   suminb|   null|
  // +---------+-------+

  spark.read.option("mergeSchema", true).parquet("data{1,2,3,4}").show()
  // +---------+-------+----+
  // |     name|   addr| sex|
  // +---------+-------+----+
  // |    cwkim| unjung|male|
  // |      sub|yangjae|male|
  // |yeonghoey| jamsil|null|
  // |   suminb|   null|male|
  // +---------+-------+----+
#+END_SRC

#+BEGIN_SRC scala
  import org.apache.spark.sql.Row
  import org.apache.spark.sql.types._
  import spark.implicits._

  val schema1 = StructType(List(StructField("x'", StringType)))
  val data1   = List(Row("yeonghoey"))
  val rdd1    = spark.sparkContext.parallelize(data1)
  val df1     = spark.createDataFrame(rdd1, schema1)
  df1.write.mode("overwrite").parquet("data1")

  val schema2 = StructType(List(StructField("x'", IntegerType)))
  val data2   = List(Row(31))
  val rdd2    = spark.sparkContext.parallelize(data2)
  val df2     = spark.createDataFrame(rdd2, schema2)
  df2.write.mode("overwrite").parquet("data2")

  // ----------------------------------------------------------------------
  
  spark.read.parquet("data{1,2}").show()
  // Caused by: java.lang.UnsupportedOperationException: Unimplemented type: StringType

  spark.read.option("mergeSchema", true).parquet("data{1,2}").show()
  // Caused by: org.apache.spark.SparkException: Failed to merge incompatible data types StringType and IntegerType
#+END_SRC

* How-to
* Links
* Staging
** TODO <2017-08-21 Mon> Required IAM Role for using S3 with Spark
- https://docs.databricks.com/user-guide/cloud-configurations/aws/iam-roles.html
  For using spark with data in s3, just s3 Programmatic Access policy required

** TODO <2017-07-21 Fri> HiveQL, Window Function
- https://cwiki.apache.org/confluence/display/Hive/LanguageManual+WindowingAndAnalytics

** TODO <2017-07-21 Fri> Retention Query
- https://blog.treasuredata.com/blog/2016/07/22/rolling-retention-done-right-in-sql/

#+BEGIN_SRC sql
  %sql
  create or replace temporary view
    RetentionByCount
  as with

  daily as (
    select distinct
      user_id,
      to_date(from_utc_timestamp(`@time`, "JST")) as day
    from UserLogins
  ),

  by_first_day as (
    select
      user_id,
      day,
      first_value(day) over (partition by user_id order by day) as first_day
    from daily
  ),

  by_diff as (
    select
      first_day,
      datediff(day, first_day) as diff
    from by_first_day
  )

  select
    first_day,
    sum(case when diff = 0 then 1 else 0 end) as day00,
    sum(case when diff = 1 then 1 else 0 end) as day01,
    sum(case when diff = 2 then 1 else 0 end) as day02,
    sum(case when diff = 3 then 1 else 0 end) as day03,
    sum(case when diff = 4 then 1 else 0 end) as day04,
    sum(case when diff = 5 then 1 else 0 end) as day05,
    sum(case when diff = 6 then 1 else 0 end) as day06,
    sum(case when diff = 7 then 1 else 0 end) as day07,
    sum(case when diff = 8 then 1 else 0 end) as day08,
    sum(case when diff = 9 then 1 else 0 end) as day09,
    sum(case when diff = 10 then 1 else 0 end) as day10,
    sum(case when diff = 11 then 1 else 0 end) as day11,
    sum(case when diff = 12 then 1 else 0 end) as day12,
    sum(case when diff = 13 then 1 else 0 end) as day13,
    sum(case when diff = 14 then 1 else 0 end) as day14,
    sum(case when diff = 15 then 1 else 0 end) as day15,
    sum(case when diff = 16 then 1 else 0 end) as day16,
    sum(case when diff = 17 then 1 else 0 end) as day17,
    sum(case when diff = 18 then 1 else 0 end) as day18,
    sum(case when diff = 19 then 1 else 0 end) as day19,
    sum(case when diff = 20 then 1 else 0 end) as day20,
    sum(case when diff = 21 then 1 else 0 end) as day21,
    sum(case when diff = 22 then 1 else 0 end) as day22,
    sum(case when diff = 23 then 1 else 0 end) as day23,
    sum(case when diff = 24 then 1 else 0 end) as day24,
    sum(case when diff = 25 then 1 else 0 end) as day25,
    sum(case when diff = 26 then 1 else 0 end) as day26,
    sum(case when diff = 27 then 1 else 0 end) as day27,
    sum(case when diff = 28 then 1 else 0 end) as day28,
    sum(case when diff = 29 then 1 else 0 end) as day29,
    sum(case when diff = 30 then 1 else 0 end) as day30
  from by_diff
  group by 1
  order by 1
#+END_SRC

#+BEGIN_SRC sql
  %sql
  create or replace temporary view 
    RetentionByPercentage
  as
  select
    first_day,
    day00 as `new`,
    round(day01 / day00 * 100, 2) as `d+1`,
    round(day02 / day00 * 100, 2) as `d+2`,
    round(day03 / day00 * 100, 2) as `d+3`,
    round(day04 / day00 * 100, 2) as `d+4`,
    round(day05 / day00 * 100, 2) as `d+5`,
    round(day06 / day00 * 100, 2) as `d+6`,
    round(day07 / day00 * 100, 2) as `d+7`,
    round(day08 / day00 * 100, 2) as `d+8`,
    round(day09 / day00 * 100, 2) as `d+9`,
    round(day10 / day00 * 100, 2) as `d+10`,
    round(day11 / day00 * 100, 2) as `d+11`,
    round(day12 / day00 * 100, 2) as `d+12`,
    round(day13 / day00 * 100, 2) as `d+13`,
    round(day14 / day00 * 100, 2) as `d+14`,
    round(day15 / day00 * 100, 2) as `d+15`,
    round(day16 / day00 * 100, 2) as `d+16`,
    round(day17 / day00 * 100, 2) as `d+17`,
    round(day18 / day00 * 100, 2) as `d+18`,
    round(day19 / day00 * 100, 2) as `d+19`,
    round(day20 / day00 * 100, 2) as `d+20`,
    round(day21 / day00 * 100, 2) as `d+21`,
    round(day22 / day00 * 100, 2) as `d+22`,
    round(day23 / day00 * 100, 2) as `d+23`,
    round(day24 / day00 * 100, 2) as `d+24`,
    round(day25 / day00 * 100, 2) as `d+25`,
    round(day26 / day00 * 100, 2) as `d+26`,
    round(day27 / day00 * 100, 2) as `d+27`,
    round(day28 / day00 * 100, 2) as `d+28`,
    round(day29 / day00 * 100, 2) as `d+29`,
    round(day30 / day00 * 100, 2) as `d+30`
  from RetentionByCount
  order by 1
#+END_SRC

{'fraction_non_alphanumeric': 0.18342911877394635, 'fraction_numerical': 0.052083333333333336, 'mean_word_length': 2.446358572312771, 'pattern_counts': {'":': 0, '<': 25, '<?xml version=': 0, '>': 33, 'https://': 16, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6048311', 'n_tokens_mistral': 13251, 'n_tokens_neox': 11934, 'n_words': 3581}