dmateusp

package main

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.SaveMode

case class User(id: Int, `type`: String)

object Main {
  val spark = SparkSession
    .builder()

dmateusp

{"lastUpload":"2020-09-02T12:11:43.678Z","extensionVersion":"v3.4.3"}

dmateusp

// load me with :load AfterRefactor.scala

type Transform = DataFrame => DataFrame

def sumAmounts(by: Column*): Transform =
  df => df.groupBy(by: _*).agg(sum(col("amount")))

def extractPayerBeneficiary(columnName: String): Transform =
  df =>
    df.withColumn(

dmateusp

// chain transform calls
dfTransactions
  .transform(extractPayerBeneficiary("details"))
  .transform(sumAmounts(date_trunc("day", col("ts")), col("details_beneficiary")))

// andThen
dfTransactions
  .transform(extractPayerBeneficiary("details") andThen sumAmounts(date_trunc("day", col("ts")), col("details_beneficiary")))

// compose

dmateusp

dfTransactions
  .transform(extractPayerBeneficiary("details", _))
  .transform(sumAmounts(_, date_trunc("day", col("ts")), col("details_beneficiary")))
  .filter(col("sum(amount)") > 25)
  .show

dmateusp

def sumAmounts(by: Column*): DataFrame => DataFrame =
  df => df.groupBy(by: _*).agg(sum(col("amount")))

def extractPayerBeneficiary(columnName: String): DataFrame => DataFrame =
  df =>
    df.withColumn(
      s"${columnName}_payer",
      regexp_extract(
        col(columnName),
        "paid by ([A-Z])",

dmateusp

sumAmounts(
  extractPayerBeneficiary(
    "details",
    dfTransactions),
  col("details_beneficiary"),
  date_trunc(
    "day",
    col("ts")
  )
).filter(col("sum(amount)") > 25).show

dmateusp

def sumAmounts(df: DataFrame, by: Column*): DataFrame =
    df.groupBy(by: _*).agg(sum(col("amount")))

def extractPayerBeneficiary(columnName: String, df: DataFrame): DataFrame =
  df.withColumn(
      s"${columnName}_payer",
      regexp_extract(
        col(columnName),
        "paid by ([A-Z])",
        1

dmateusp

case class Transaction(details: String, amount: Int, ts: Timestamp)

val dfTransactions: DataFrame = Seq(
    Transaction("paid by A to X", 100, Timestamp.valueOf("2018-01-05 08:00:00")),
    Transaction("paid by B to X", 10, Timestamp.valueOf("2018-01-05 11:00:00")),
    Transaction("paid by C to Y", 15, Timestamp.valueOf("2018-01-06 12:00:00")),
    Transaction("paid by D to Z", 50, Timestamp.valueOf("2018-01-06 15:00:00"))
).toDF

dmateusp

-- run: stack ghci --package avro --package text --package bytestring --ghc-options -XOverloadedStrings
-- :load AvroLearn.hs
{-# LANGUAGE OverloadedStrings #-}

import           Control.Monad        ((<=<))
import           Data.Avro            (FromAvro (..), HasAvroSchema (..),
                                       Schema, ToAvro (..), badValue,
                                       decodeContainer, encodeContainer, record,
                                       (.:), (.=))
import qualified Data.Avro.Schema     as S (Field (..), Order (..), Result (..),