Don’t Rip and Replace Your Platform. Augment with Standards.

↳ Arrow Flight Software Stack

↳ Under the Hood: How Flight Sends Arrow Data

Copy
  import ibis
  from ibis import _
  from pyspark.sql import SparkSession

  def build_features(table):
    user_logs = table.mutate(log_date =
  _.date.cast("string").to_timestamp("%Y%m%d").date()).drop("date")
    user_logs_agg = user_logs.aggregate(
      by=["msno"],
      sum_total_secs = _.total_secs.sum(),
      avg_total_secs = _.total_secs.mean(),
      max_total_secs = _.total_secs.max(),
      min_total_secs = _.total_secs.min(),
      total_days_active = _.count(),
      first_session = _.log_date.max(),
      most_recent_session = _.log_date.min(),
      total_songs_listened = (_.num_25 + _.num_50 + _.num_75 + _.num_985 + _.num_100).sum(),
      avg_num_songs_listened = (_.num_25 + _.num_50 + _.num_75 + _.num_985 + _.num_100).mean(),
      percent_unique = (_.num_unq/(_.num_25 + _.num_50 + _.num_75 + _.num_985 + _.num_100)).mean(),)

  return user_logs_agg

  session =
  SparkSession.builder.appName("kkbox-customer-churn").getOrCreate
  #if you don’t use legacy mode here, you’ll need to change the timestamp formatting code
  session.conf.set("spark.sql.legacy.timeParserPolicy","LEGACY")
  df =
  session.read.parquet("gs://voltrondata-demo-data/kkbox-churn/user_logs/*.parquet")
  df.createOrReplaceTempView("user_logs")

  ()ibis_con = ibis.pyspark.connect(session)
  table = ibis_con.table("user_logs")

  user_logs_agg = build_features(table)
  ibis_con.create_table("user_logs_agg", user_logs_agg)

↳ Connecting Ibis to distributed systems