1. Load the model from the blob store.
2. Transform the input DataFrame (after the Scala transform but before the SQL transform).
3. Add the predicted output field (specified in the model metadata stored in the blob store) to the projected fields list.
This lets you use Spark ML models to make predictions in a more scalable, performant manner than what can be achieved with a Machine Learning index stage. |
Function for `transformScala`:
```scala theme={"dark"}
def transform(inputDF: Dataset[Row]) : Dataset[Row] = {
{/* // do transformations and/or filter the inputDF here */}
}
Your script can rely on the following vals:
spark: SparkSession
sc: SparkContext
fusionZKConn: ZKConnection // needed to access Fusion API
solrCollection: SolrCollection // output collection
jobId: Loader job config ID
Also, the following classes have already been imported:
import org.apache.spark.SparkContext._
import spark.implicits._
import spark.sql
import org.apache.spark.sql.functions._
import com.lucidworks.spark.util.{SparkShellSupport => _lw}
import com.lucidworks.spark.job.sql.SparkSQLLoader
import com.lucidworks.spark.ml.recommenders.SolrCollection
import com.lucidworks.spark.ZKConnection
import org.apache.spark.sql.{Dataset, Row}
```
### Output settings
| Setting | Description |
| ------------------------------ | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `outputCollection` | Name of the Fusion collection to write to. The Parallel Bulk Loader uses the Collections API to resolve the underlying Solr collection at runtime. |
| `outputIndexPipeline` | Name of a Fusion index pipeline to which to send documents, instead of directly indexing to Solr. This option lets you perform additional ETL (extract, transform, and load) processing on the documents before they are indexed in Solr. If you need to write to time-partitioned indexes, then you must use an index pipeline, because writing directly to Solr is not partition aware. |
| `defineFieldsUsingInputSchema` | Flag to indicate if the Parallel Bulk Loader should use the input schema to create fields in Solr, after applying the Scala and/or SQL transformations. If `false`, then the Parallel Bulk Loader relies on the Fusion index pipeline and/or Solr field guessing to create the fields. If `true`, only fields that do not exist already in Solr are created. Consequently, if there is a type mismatch between an existing field in Solr and the input schema, you will need to use a transformation to rename the field in the input schema. |
| `clearDatasource` | If checked, the Parallel Bulk Loader deletes any existing documents in the output collection that match the query `_lw_loader_id_s:
|
| `outputPartitions` | Coalesce the DataFrame into N partitions before writing to Solr. This can help spread the indexing work out across more executors that are available in Spark, or limit the parallelism when writing to Solr. |
## Tune performance
As the name of the Parallel Bulk Loader job implies, it is designed to ingest large amounts of data into Fusion by parallelizing the work across your Spark cluster. To achieve scalability, you might need to increase the amount of memory and/or CPU resources allocated to the job.
For Spark configuration information, see Spark operations.
You can pass these properties in the job configuration to override the default Spark shell options:
| Parameter Name | Description and Default |
| ------------------------ | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `--driver-cores` | Cores for the driver Default: `1` |
| `--driver-memory` | Memory for the driver (for example, `1000M` or `2G`) Default: `1024M` |
| `--executor-cores` | Cores per executor Default: 1 in YARN mode, or all available cores on the worker in standalone mode |
| `--executor-memory` | Memory per executor (for example, `1000M` or `2G`) Default: `1G` |
| `--total-executor-cores` | Total cores for all executors Default: Without setting this parameter, the total cores for all executors is the number of executors in YARN mode, or all available cores on all workers in standalone mode. |
## Examples
Here we provide screenshots and example JSON job definitions to illustrate key points about how to load from different data sources.
### Use NLP during indexing
In this example, we leverage the John Snow labs NLP library during indexing. This is just quick-and-dirty to show the concept.
Also see:
* [https://github.com/JohnSnowLabs/spark-nlp](https://github.com/JohnSnowLabs/spark-nlp)
* [https://databricks.com/blog/2017/10/19/introducing-natural-language-processing-library-apache-spark.html](https://databricks.com/blog/2017/10/19/introducing-natural-language-processing-library-apache-spark.html)
Use this transform Scala script:
```scala theme={"dark"}
import com.johnsnowlabs.nlp._
import com.johnsnowlabs.nlp.annotators._
import org.apache.spark.ml.Pipeline
import com.johnsnowlabs.nlp.annotators.sbd.pragmatic.SentenceDetector
def transform(inputDF: Dataset[Row]) : Dataset[Row] = {
val documentAssembler = new DocumentAssembler().setInputCol("plot_txt_en").setOutputCol("document")
val sentenceDetector = new SentenceDetector().setInputCols(Array("document")).setOutputCol("sentences")
val finisher = new Finisher()
.setInputCols("sentences")
.setOutputCols("sentences_ss")
.setOutputAsArray(true)
.setCleanAnnotations(false)
val pipeline = new Pipeline().setStages(Array(documentAssembler,sentenceDetector,finisher))
pipeline.fit(inputDF).transform(inputDF).drop("document").drop("sentences")
}
```
Be sure to add the `JohnSnowLabs:spark-nlp:1.4.2` package using Spark Shell Options.
### Clean up data with SQL transformations
Fusion has a Local Filesystem connector that can handle files such as CSV and JSON files. Using the Parallel Bulk Loader lets you leverage features that are not in the Local Filesystem connector, such as using SQL to clean up the input data.
Use the following SQL to clean up the input data before indexing:
```sql theme={"dark"}
SELECT _c0 as user_id,
CAST(_c1 as INT) as age,
_c2 as gender,
_c3 as occupation,
_c4 as zip_code
FROM _input
Job JSON:
{
"type" : "parallel-bulk-loader",
"id" : "csv",
"format" : "csv",
"path" : "/Users/tjp/dev/lw/projects/fusion-spark-bootcamp/labs/movielens/ml-100k/u.user",
"readOptions" : [ {
"key" : "delimiter",
"value" : "|"
}, {
"key" : "header",
"value" : "false"
} ],
"outputCollection" : "users",
"clearDatasource" : false,
"defineFieldsUsingInputSchema" : true,
"atomicUpdates" : false,
"transformSql" : "SELECT _c0 as user_id, \n CAST(_c1 as INT) as age, \n _c2 as gender,\n _c3 as occupation,\n _c4 as zip_code \n FROM _input"
}
```
### Read from S3
It is easy to read from an S3 bucket without pulling data down to your local workstation first. To avoid exposing your AWS credentials, add them to a file named `core-site.xml` in the `apps/spark-dist/conf` directory, such as:
```xml theme={"dark"}
You will need to add the `org.apache.hadoop:hadoop-aws:2.7.3` package to the job using the `--packages` Spark option. Also, you will need to exclude the `com.fasterxml.jackson.core:jackson-core,joda-time:joda-time` packages using the `--exclude-packages` option.
You can also read from Google Cloud Storage (GCS), but you will need a few more properties in your `core-site.xml`; see [Installing the Cloud Storage connector](https://cloud.google.com/dataproc/docs/concepts/connectors/install-storage-connector).
### Read from Parquet
Reading from parquet files is built into Spark using the "parquet" format. For additional read options, see
[Configuration of Parquet](https://spark.apache.org/docs/latest/sql-programming-guide.html#configuration).
Job JSON:
```json theme={"dark"}
{
"type" : "parallel-bulk-loader",
"id" : "ecomm demo parquet signals",
"format" : "parquet",
"path" : "./part-00000-c1951958-98ae-4f2a-b7b4-2e3a69fcf403-c000.snappy.parquet",
"outputCollection" : "best-buy_signals",
"clearDatasource" : false,
"defineFieldsUsingInputSchema" : true,
"atomicUpdates" : false
}
```
This example also uses the `transformScala` option to filter and transform the input DataFrame into a better form for indexing using the following Scala script:
```scala theme={"dark"}
import java.util.Calendar
import java.util.Locale
import java.util.TimeZone
def transform(inputDF: Dataset[Row]) : Dataset[Row] = {
{/* // do transformations and/or filter the inputDF here */}
val signalsDF =
inputDF.filter((unix_timestamp($"timestamp_tdt", "MM/dd/yyyy HH:mm:ss.SSS") < 1325376000))
val now = System.currentTimeMillis()
val maxDate = signalsDF.agg(max("timestamp_tdt")).take(1)(0).getAs[java.sql.Timestamp](0).getTime
val diff = now - maxDate
val add_time =
udf((t: java.sql.Timestamp, diff : Long) => new java.sql.Timestamp(t.getTime + diff))
val day_of_week = udf((t: java.sql.Timestamp) => {
val calendar = Calendar.getInstance(TimeZone.getTimeZone("UTC"))
calendar.setTimeInMillis(t.getTime)
calendar.getDisplayName(Calendar.DAY_OF_WEEK, Calendar.LONG, Locale.getDefault)
})
{/* //Remap some columns to bring the timestamps current */}
signalsDF
.withColumnRenamed("timestamp_tdt", "orig_timestamp_tdt").withColumn("timestamp_tdt", add_time($"orig_timestamp_tdt", lit(diff)))
.withColumn("date", $"timestamp_tdt")
.withColumn("tx_timestamp_txt", date_format($"timestamp_tdt", "E YYYY-MM-d HH:mm:ss.SSS Z"))
.withColumn("param.query_time_dt", $"timestamp_tdt")
.withColumn("date_day", date_format(date_sub($"date", 0), "YYYY-MM-d'T'HH:mm:ss.SSS'Z'"))
.withColumn("date_month", date_format(trunc($"date", "mm"), "YYYY-MM-d'T'HH:mm:ss.SSS'Z'"))
.withColumn("date_year", date_format(trunc($"date", "yyyy"), "YYYY-MM-d'T'HH:mm:ss.SSS'Z'"))
.withColumn("day_of_week", day_of_week($"date"))
}
```
### Read from JDBC tables
You can use the Parallel Bulk Loader to parallelize reads from JDBC tables, if the tables have numeric columns that can be partitioned into relatively equal partition sizes. In the example below, we partition the employees table into 4 partitions using the `emp_no` column (`int`). Behind the scenes, Spark sends four separate queries to the database and processes the result sets in parallel.
#### Load the JDBC driver JAR file into the Blob store
Before you ingest from a JDBC data source, you need to use the Fusion Admin UI to upload the JDBC driver JAR file into the blob store.
Alternatively, you can add the JAR file to the Fusion blob store with `resourceType=spark:jar`; for example:
```bash wrap theme={"dark"}
curl -XPUT -H "Content-type:application/octet-stream" "http://
For more information on reading from JDBC-compliant databases, see:
* [http://spark.apache.org/docs/latest/sql-programming-guide.html#jdbc-to-other-databases](http://spark.apache.org/docs/latest/sql-programming-guide.html#jdbc-to-other-databases)
* [https://medium.com/@radek.strnad/tips-for-using-jdbc-in-apache-spark-sql-396ea7b2e3d3](https://medium.com/@radek.strnad/tips-for-using-jdbc-in-apache-spark-sql-396ea7b2e3d3)
```json theme={"dark"}
{
"type" : "parallel-bulk-loader",
"id" : "load dbtable",
"format" : "jdbc",
"readOptions" : [ {
"key" : "url",
"value" : "jdbc:mysql://localhost/employees?user=?&password=?"
}, {
"key" : "dbtable",
"value" : "employees"
}, {
"key" : "partitionColumn",
"value" : "emp_no"
}, {
"key" : "numPartitions",
"value" : "4"
}, {
"key" : "driver",
"value" : "com.mysql.jdbc.Driver"
}, {
"key" : "lowerBound",
"value" : "$MIN(emp_no)"
}, {
"key" : "upperBound",
"value" : "$MAX(emp_no)"
} ],
"outputCollection" : "employees",
"clearDatasource" : false,
"defineFieldsUsingInputSchema" : true,
"atomicUpdates" : false
}
```
Notice the use of the `$MIN(emp_no)` and `$MAX(emp_no)` macros in the read options. These are macros offered by the Parallel Bulk Loader to help configure parallel reads of JDBC tables. Behind the scenes, the macros are translated into SQL queries to get the MAX and MIN values of the specified field, which Spark uses to compute splits for partitioned queries. As mentioned above, the field must be numeric and must have a relatively balanced distribution of values between MAX and MIN; otherwise, you are unlikely to see much performance benefit to partitioning.
### Index HBase tables
To index an HBase table, use the [Hortonworks connector](https://github.com/hortonworks-spark/shc).
Notice the use of the `$lastTimestamp` macro in the read options. This lets us filter rows read from HBase using the timestamp of the last document the Parallel Bulk Loader wrote to Solr, that is, to get the newest updates from HBase only (incremental updates). Most Spark data sources provide a way to filter results based on timestamp.
Job JSON:
```json theme={"dark"}
{
"type" : "parallel-bulk-loader",
"id" : "hbase",
"format" : "org.apache.spark.sql.execution.datasources.hbase",
"readOptions" : [ {
"key" : "catalog",
"value" : "{ \"table\":{\"namespace\":\"default\", \"name\":\"fusion_nums\"}, \"rowkey\":\"key\", \"columns\":{ \"id\":{\"cf\":\"rowkey\", \"col\":\"key\", \"type\":\"string\"}, \"lw_c1_s\":{\"cf\":\"lw\", \"col\":\"c1\", \"type\":\"string\"}, \"lw_c2_s\":{\"cf\":\"lw\", \"col\":\"c2\", \"type\":\"string\"} } }"
}, {
"key" : "minStamp",
"value" : "$lastTimestamp(EPOCH_MS)"
} ],
"outputCollection" : "hbase",
"timestampFieldName" : "timestamp_tdt",
"clearDatasource" : false,
"defineFieldsUsingInputSchema" : true,
"atomicUpdates" : false,
"shellOptions" : [ {
"key" : "--packages",
"value" : "com.hortonworks:shc-core:1.1.1-2.1-s_2.11"
}, {
"key" : "--repositories",
"value" : "https://mvnrepository.com/repos/hortonworks-releases"
} ]
}
```
### Index Elastic data
With Elasticsearch 6.2.2 using the `org.elasticsearch:elasticsearch-spark-20_2.11:6.2.1` package, here is a Scala script to run in `bin/spark-shell` to index some test data:
```scala theme={"dark"}
import spark.implicits._
case class SimpsonCharacter(name: String, actor: String, episodeDebut: String)
val simpsonsDF = sc.parallelize(
SimpsonCharacter("Homer", "Dan Castellaneta", "Good Night") ::
SimpsonCharacter("Marge", "Julie Kavner", "Good Night") ::
SimpsonCharacter("Bart", "Nancy Cartwright", "Good Night") ::
SimpsonCharacter("Lisa", "Yeardley Smith", "Good Night") ::
SimpsonCharacter("Maggie", "Liz Georges and more", "Good Night") ::
SimpsonCharacter("Sideshow Bob", "Kelsey Grammer", "The Telltale Head") :: Nil).toDF()
val writeOpts = Map("es.nodes" -> "127.0.0.1",
"es.port" -> "9200",
"es.index.auto.create" -> "true",
"es.resouce.auto.create" -> "shows/simpsons")
simpsonsDF.write.format("org.elasticsearch.spark.sql").mode("Overwrite").save("shows/simpsons")
```
Job JSON:
```json theme={"dark"}
{
"type" : "parallel-bulk-loader",
"id" : "elastic",
"format" : "org.elasticsearch.spark.sql",
"readOptions" : [ {
"key" : "es.nodes",
"value" : "127.0.0.1"
}, {
"key" : "es.port",
"value" : "9200"
}, {
"key" : "es.resource",
"value" : "shows/simpsons"
} ],
"outputCollection" : "hbase_signals_aggr",
"clearDatasource" : false,
"defineFieldsUsingInputSchema" : true,
"atomicUpdates" : false,
"shellOptions" : [ {
"key" : "--packages",
"value" : "org.elasticsearch:elasticsearch-spark-20_2.11:6.2.2"
} ]
}
```
### Read from Couchbase
To index a Couchbase bucket, use the official Couchbase Spark connector found [here](https://spark-packages.org/package/couchbase/couchbase-spark-connector).
For example, we will create a test bucket in Couchbase. If you already have a bucket in Couchbase, feel free to use that and skip to the test data setup section. This test was performed using Couchbase Server 6.0.0.
1. Create a bucket test in the Couchbase admin UI. Give access to a system account user to use in the Parallel Bulk Loader job config.
2. Connect to Couchbase using the command line client cbq. For example, `cbq -e=http://
### XML setup
XML is a supported format that requires settings for `format` and `--packages`. In addition, you must specify the filepath in the `readOptions` section. For example:
```json theme={"dark"}
{
"type": "parallel-bulk-loader",
"id": "pbl",
"format": "com.databricks.spark.xml",
"path": "this-is-ignored",
"readOptions": [
{
"key": "rowTag",
"value": "tag_name_representing_record"
},
{
"key": "path",
"value": "/home/user/file.xml.gz"
}
],
"outputCollection": "output_collection_name",
"clearDatasource": true,
"defineFieldsUsingInputSchema": true,
"atomicUpdates": false,
"transformScala": "",
"shellOptions": [
{
"key": "--packages",
"value": "com.databricks:spark-xml_2.11:0.5.0"
}
],
"cacheAfterRead": false
}
```