# Code Examples

### MongoDB

> Read from database

```
import com.mongodb.spark._
import com.mongodb.spark.sql._
import com.mongodb.spark.config._

val dfFA = spark.read.option("database", "foxconn-analytics")
                     .option("collection", "collect_pageview")
                     // option uri can be omitted if it has been defined in Spark interpreter 
                     .option("uri", "mongodb://mongo201,mongo202,mongo203/?replicaSet=bigdata&readPreference=secondaryPreferred")
                     .mongo()
```

> Read from database with schema specifying by case class

```
case class EHAVE_TBL(
    CNAME: String, CCOVEY: String, ISCORE: String,
    IFPASS: String, FACTORY: String, ISEX: String
)

val dsEHAVE = spark.read
                   .option("database", "IEDB")
                   .option("collection", "EHAVE_TBL_2016")
                   .mongo[EHAVE_TBL]()
```

> Write to database

```
MongoSpark.save(dfStaff2Mongo.write.format("com.mongodb.spark.sql")
                     .options(Map("database" -> "HR", "collection" -> "EMP_TAGS_20170324"))
                     .mode("overwrite")
               )
```

### MySQL/Maridb

> Read from database

```
val dfLog = spark.read.format("jdbc")
                 .options(
                     Map("url" -> "jdbc:mysql://192.168.2.52/IIS?user=ACCOUNT_NAME&password=PASSWORD",
                         "driver" -> "com.mysql.jdbc.Driver",
                         "dbtable" -> "IIS.TV",
                         "partitionColumn" -> "id",
                         "lowerBound" -> "1",
                         "upperBound" -> "50000000",
                         "numPartitions" -> "200"))
                 .load()
                 .cache()
```

> Write to database

```
val prop = new java.util.Properties
prop.setProperty("driver", "com.mysql.jdbc.Driver")
prop.setProperty("user", "<Username>")
prop.setProperty("password", "<Password>") 

dfDiff.write
      .mode("overwrite")
      .jdbc(jdbc:mysql://192.168.2.52/<Database Name>?&useUnicode=yes&characterEncoding=UTF-8, "<Table Name>", prop)
```

### Orc File

> Read from database

```
val df = spark.read.orc("/orc/PromoterLearn.orc")
```

> Write to database

```
df.select('JobNo, 'Name, 'Sex)
  .write
  .mode("overwrite")
  .orc("/orc/PromoterLearn.orc")
```

### Parquet

> Read from database

```
val df = spark.read.parquet("/parquet/PromoterLearn.parquet")
```

> Write to database

```
dfTaobao.select('JobNo,'Name,'Sex)
        .write
        .mode("overwrite")
        .parquet("/parquet/PromoterTaobao2017.parquet")
```

### CSV

> Read from CSV

```
val dsProvTest = spark.read
                      .option("header", true)
                      .csv("/csv/ProvPlayTest.csv")
```

> Write to CSV

```
dsProvPlay.write
          .mode("overwrite")
          .csv("/csv/ProvPlayTest.csv")
```

### Oracle

> Read from database

```
val dfDL = spark.read.format("jdbc")
                .options(
                    Map("url" -> "jdbc:oracle:thin:<AccountName>/<Password>@<DB's IP>:<DB's Port>:<SID Name>",
                        "driver" -> "oracle.jdbc.driver.OracleDriver",
                        "dbtable" -> "DL_TBL"))
                .load()
```

> Write to database

```
```

### MS SQL Server

> Read from database

```
val dfSQLServer = spark.read.format("jdbc")
                       .option("url","jdbc:sqlserver://<DB's IP>:1433;database=educ2013;user=<Username>;password=<Password>")
                       .option("driver", "com.microsoft.sqlserver.jdbc.SQLServerDriver")
                       .option("dbtable", "user_tbl")
                       .option("partitionColumn", "IID")
                       .option("lowerBound", "1")
                       .option("upperBound", "3370000")
                       .option("numPartitions", "100")
                       .load()
```

## Use isin

```
val lstReceiverName = dfTaobao.groupBy('ReceiverName)
                              .count()
                              .orderBy('count.desc)
                              .filter('count >= 10)
                              .agg(collect_list($"ReceiverName").alias("RName"))
                              .first()
                              .getSeq(0)

dfTaobao.filter('ReceiverName.isin(lstReceiverName:_*))
        .groupBy('ReceiverName)
        .pivot("GoodsID")
        .count()
        .na.fill(0)
        .withColumn("SubTotal", when('GoodsID.like("LCD-%"), lit("平板電視"))
        .groupBy().sum()
        .printCSV
```

## Concatenate all columns

```
concat(dsLiveEPG.columns.map(c => col(c)):_*)
```

## Hash an array of columns

```
import sqlContext.implicits._
import org.apache.spark.sql.functions.udf
def hasher(data: AnyRef *) = (data.map(_.hashCode).sum % 10)
val getBucket = udf(hasher _)
val df = sc.parallelize(('a' to 'z').map(_.toString) zip (1 to 30)).toDF("c1","c2")
df.withColumn("bucket", getBucket(array(df.columns.map(df.apply): _*))).show()
```

## UDF

**User-Defined Functions**

***UDF*** is a feature of Spark SQL to define new Column-based functions that extend the vocabulary of Spark SQL’s DSL for transforming Datasets.

```
val replaceWrongDevice = udf {(device: String) => 
  if(device == "ANOIRD" || device == "ANDOIRD") "ANDROID" else device
}
```

```
val createUUID3 = udf { x: String => UUID.nameUUIDFromBytes(x.getBytes).toString }
val createUUID4 = udf { () => java.util.UUID.randomUUID().toString }
```

## converts a Scala list to its Java equivalent

```
import scala.collection.JavaConverters._
import org.bson.Document

val documents = sc.parallelize(
  Seq(new Document("fruits", List("apples", "oranges", "pears").asJava))
)
MongoSpark.save(documents)
```

## Window Function

```
import org.apache.spark.sql.functions.rank
import org.apache.spark.sql.expressions.Window

val df = spark.createDataFrame(Seq(
  (1, "philip", 2.0, "montreal"),
  (2, "john", 4.0, "montreal"),
  (3, "charles", 2.0, "texas"))).toDF("Id", "username", "rating", "city")

val w = Window.orderBy($"city")
df.withColumn("id", rank().over(w)).show()
```

```
from pyspark.sql.functions import UserDefinedFunction
from pyspark.sql.types import StringType

name = 'target_column'
udf = UserDefinedFunction(lambda x: 'new_value', Stringtype())
new_df = old_df.select(*[udf(column).alias(name) if column == name else column for column in old_df.columns])
```

## Rename Dataset Column Name

```
val newNames = Seq("datetime", "userid", "URL", "drop")
val dfNewOne = dsUserWeb.toDF(newNames: _*)
```

## Regular Expression to extract values

```
val dfTaobao = df0207.withColumn("GoodsID",
    regexp_extract('GoodsTitle, """((?:LCD|SJ|BCD)\-[a-zA-Z0-9/\-]*).*""", 1))
    regexp_extract(ReceiverAddress, '([^\\\\s]*)\\\\s+([^\\\\s]*)\\\\s+.*', 1))
```

## Transpose rows to column

```
case class Tag(ID: String, TagName: String, TagValue: String)
var dsTag = Seq(Tag("1","A","1"),Tag("1","B","2"),Tag("2","A","2"),Tag("3","B","2")).toDS()
dsTag.groupBy('ID).pivot("TagName").agg(collect_list($"TagValue").getItem(0)).show
```