Parallel Bulk Loader - Lucidworks documentation

The Parallel Bulk Loader (PBL) job enables bulk ingestion of structured and semi-structured data from big data systems, NoSQL databases, and common file formats like Parquet and Avro. Datasources the PBL uses include not only common file formats, but Solr databases, JDBC-compliant databases, MongoDB databases and more. In addition, the PBL distributes the load across the Lucidworks Search Spark cluster to optimize performance. And because no parsing is needed, indexing performance is also maximized by writing directly to Solr.

Usage

Use this job to load data into Lucidworks Search from a SparkSQL-compliant datasource, and then send the data to any Spark-supported datasource such as Solr, an index pipeline, etc. To create a Parallel Bulk Loader job, sign in to Lucidworks Search and click Collections > Jobs. Then click Add+ and in the Custom and Others Jobs section, select Parallel Bulk Loader. You can enter basic and advanced parameters to configure the job. If the field has a default value, it is populated when you click to add the job. Parallel Bulk Loader can be configured for many different use cases. Two examples are:

Organizations that need to meet financial/bank-level transactional integrity requirements use SQL databases. Those datasources are structured in relational tables and ensure data integrity even if errors or power failures occur. In addition, these datasources are useful in large-scale operations that employ complex queries for analytics and reporting. For example, if categories and products have various prices, discounts, and offering dates, a SQL datasource is the most efficient option. Lucidworks supports SQL databases such as JDBC-compliant databases.
In contrast, NoSQL databases are based on documents and allow for more flexibility with structured, semi-structured, and unstructured data. For example, your organization might need information about user product reviews or session data. And if your organization needs to process massive amount of data from multiple systems, NoSQL is an efficient option. Lucidworks supports NoSQL databases such as MongoDB and Apache Cassandra.

Example configuration

The following is an example configuration. The table after the configuration defines the fields.

{
        "id": "store_typeahead_entity_load",
        "format": "solr",
        "path": "https://example.com/products_database/*.*",
        "streaming": [
            {
                "enableStreaming": true,
                "outputMode": "append"
            }
        ],
        "readOptions": [
            {
                "key": "collection",
                "value": "store"
            }
        ],
        "outputCollection": "store_typeahead",
        "outputIndexPipeline": "store_typeahead",
        "outputParser": "store_typeahead",
        "clearDatasource": true,
        "outputPartitions": 5,
        "optimizeOutput": 2,
        "defineFieldsUsingInputSchema": true,
        "atomicUpdates": false,
        "writeOptions": [
            {
                "key": "write_collection",
                "value": "store2"
            }
        ],
        "transformScala": "import script",
        "mlModelId": "llm_model_id",
        "sparkConfig": [
            {
                "key": "spark.sql.caseSensitive",
                "value": "true"
            },
            {
                "key": "spark.typeField_1",
                "value": "Suggested_Doc, doc_title, url, doc_locale, doc_sector_ss, doc_economic_buyer_ss, doc_topic_ss, title_s"
            }
        ],
        "cacheAfterRead": false,
        "continueAfterFailure": false,
        "type": "parallel-bulk-loader",
        "updates": [
            {
                "userId": "service_account",
                "timestamp": "2024-05-06T09:06:43.739877Z"
            },
            {
                "userId": "power_user",
                "timestamp": "2024-07-30T20:30:31.347930292Z"
            },
            {
                "userId": "power_user",
                "timestamp": "2024-07-30T20:30:31.350243642Z"
            }
        ]
    }

Additional field information

This section provides more detailed information about some of the configuration fields.

format. Spark scans the job’s classpath for a class named DefaultSource in the <format> package. For the solr format, where the solr.DefaultSource class is defined in the spark-solr repository.

transformScala. If the Scala script is not sufficient, you might need the full power of the Spark API to transform data into an indexable form.

The transformScala option lets you filter and/or transform the input DataFrame any way you would like. You can even define UDFs to use during your transformation. For an example of using Scala to transform the input DataFrame before indexing in Solr, see the Import with the Bulk Loader example.
Another powerful use of the transformScala option is that you can pull in advanced libraries, such as Spark NLP (from John Snow Labs) to do NLP work on your content before indexing. See the Import with the Bulk Loader example.

Your Scala script can do other things but, at a minimum, it must define the following function that the Parallel Bulk Loader invokes:

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}

cacheAfterRead. This hidden field specifies if input data is retained in memory (and on disk as needed) after reading. If set to true, it may help stability of the job by reading all data from the input source first before transforming or writing to Solr. This could make the job run slower because it adds an intermediate write operation. For example, false.
updates. This field lists the userId accounts that have been updated. The timestamp contains the date and time the account was updated. The value is displayed in Unix epoch time in a yyyy-mm-ddThh:mm:ssZ format. For example, the power_user was updated on 2024-07-30T20:30:31.350243642Z.

Import with the Bulk Loader

Create and run Parallel Bulk Loader jobs

Use the Jobs manager to create and run Parallel Bulk Loader jobs. You can also use the Scheduler to schedule jobs.In the procedures, select Parallel Bulk Loader as the job type and configure the job using the following information.

Configuration settings for the Parallel Bulk Loader job

This section provides configuration settings for the Parallel Bulk Loader job. Also see configuration properties in the Configuration reference.

Read settings

Setting	Description
`format`	Unique identifier of the data source provider. Spark scans the job’s `classpath` for a class named `DefaultSource` in the `<format>` package. For example, for the `solr` format, we provide the `solr.DefaultSource` class in our `spark-solr` repository:
`path` (optional)	Comma-delimited list of paths to load. Some data sources, such as parquet, require a path. Others, such as Solr, do not. Refer to the documentation for your data source to determine if you need to provide a path.
`readOptions`	Options passed to the Spark SQL data source to configure the read operation. Options differ for every data source. Refer to the specific data source documentation for more information.
`sparkConfig` (optional)	List of Spark configuration settings needed to run the Parallel Bulk Loader.
`shellOptions`	Behind the scenes, the Parallel Bulk Loader job submits a Scala script to the Fusion Spark shell. The `shellOptions` setting lets you pass any additional options needed by the Spark shell. The two most common options are `--packages` and `--repositories`: `--packages` Comma-separated list of Maven coordinates of JAR files to include on the driver and executor classpaths. Spark searches the local Maven repository, and then Maven central and any additional remote repositories given in the config. The format for the coordinates should be `groupId:artifactId:version`. The HBase example below demonstrates the use of the packages option for loading the `com.hortonworks:shc-core:1.1.1-2.1-s_2.11` package. TIP: Use the `https://spark-packages.org/` site to find interesting packages to add to your Parallel Bulk Loader jobs. `--repositories` Comma-separated list of additional remote Maven repositories to search for the Maven coordinates given in the `packages` config setting. The Index HBase tables example below demonstrates the use of the `repositories` option for loading the `com.hortonworks:shc-core:1.1.1-2.1-s_2.11` package from this repository.
`timestampFieldName`	For datasources that support time-based filters, the Parallel Bulk Loader computes the timestamp of the last document written to Solr and the current timestamp of the Parallel Bulk Loader job. For example, the HBase data source lets you filter the read between a `MIN_STAMP` and `MAX_STAMP`, for example: `val timeRangeOpts = Map(HBaseRelation.MIN_STAMP -> minStamp.toString, HBaseRelation.MAX_STAMP -> maxStamp.toString)` lets Parallel Bulk Loader jobs run on schedules, and pull only the newest rows from the underlying datasources. To support timestamp based filtering, the Parallel Bulk Loader provides two simple macros: `$lastTimestamp(format)` `$nowTimestamp(format)` The `format` argument is optional. If not supplied, then an ISO-8601 date/time string is used. The `timestampFieldName` setting is used to determine the value of `lastTimestamp`, using a Top 1 query to Solr to get the max timestamp. You can also pass `$lastTimestamp(EPOCH)` or `$lastTimestamp(EPOCH_MS)` to get the timestamp in seconds or milliseconds. See the Index HBase tables example below for an example of using this configuration property.

Transformation settings

Setting	Description
`transformScala`	Sometimes, you can write a small script to transform input data into the correct form for indexing. But at other times, you might need the full power of the Spark API to transform data into an indexable form. The `transformScala` option lets you filter and/or transform the input DataFrame any way you would like. You can even define UDFs to use during your transformation. For an example of using Scala to transform the input DataFrame before indexing in Solr, see the Read from Parquet example. Another powerful use of the `transformScala` option is that you can pull in advanced libraries, such as Spark NLP (from John Snow Labs) to do NLP work on your content before indexing. See the Use NLP during indexing example. Your Scala script can do other things but, at a minimum, it must define the function that the Parallel Bulk Loader invokes (see below this table).
`transformSql`	The `transformSql` option lets you write a SQL query to transform the input DataFrame. The SQL is executed after the `transformScala` script (if both are defined). The input DataFrame is exposed to your SQL as the `_input` view. See the Clean up data with SQL transformations example below for an example of using SQL to transform the input before indexing in Solr. This option also lets you leverage the UDF/UDAF functions provided by Spark SQL.
`mlModelId`	If you have a Spark ML PipelineModel loaded into the blob store, you can supply the blob ID to the Parallel Bulk Loader and it will: 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:

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:<JOB>`. Consequently, the Parallel Bulk Loader adds two metadata fields to each row: `_lw_loader_id_s` and `_lw_loader_job_id_s`.
`atomicUpdates`	Flag to send documents directly to Solr as atomic updates instead of as new documents. This option is not supported when using an index profile. Also note that the Parallel Bulk Loader tracking fields `_lw_loader_id_s` and `_lw_loader_job_id_s` are not sent when using atomic updates, so the clear datasource option does not work with documents created using atomic updates.
`outputOptions`	Options used when writing directly to Solr. See Spark-Solr: https://github.com/lucidworks/spark-solr#index-parameters For example, if your docs are relatively small, you might want to increase the `batch_size` (2000 default) as shown below:
`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:

Use this transform Scala script:

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:

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:

<configuration>
  <property>
    <name>fs.s3a.access.key</name>
    <value>???</value>
  </property>
  <property>
    <name>fs.s3a.secret.key</name>
    <value>???</value>
  </property>
</configuration>

Then you can load files using the S3a protocol, such as: s3a://sstk-dev/data/u.user. If you are running a Fusion cluster then each instance of Fusion will need a core-site.xml file. S3a is the preferred protocol for reading data into Spark because it uses Amazon’s libraries to read from S3 instead of the legacy Hadoop libraries. If you need other S3 protocols (for example, s3 or s3n) you will need to add the equivalent properties to core-site.xml.

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.

Read from Parquet

Reading from parquet files is built into Spark using the “parquet” format. For additional read options, see Configuration of Parquet.

Job JSON:

{
  "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:

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:

curl -XPUT -H "Content-type:application/octet-stream" "http://<FUSION_HOST>/api/blobs/mysql_jdbc_jar?resourceType=spark:jar" --data-binary @mysql-connector-java-5.1.45-bin.jar

At runtime, Fusion’s Spark job management framework knows how to add any JAR files with resourceType=spark:jar from the blob store to the appropriate classpaths before running a Parallel Bulk Loader job.

Read from a table

For more information on reading from JDBC-compliant databases, see:

{
  "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.

The Parallel Bulk Loader lets us replace the HBase Indexer.

You will need to add an hbase-site.xml (and possibly core-site.xml) to apps/spark-dist/conf in Fusion, for example:

<configuration>
 <property>
    <name>hbase.defaults.for.version.skip</name>
    <value>true</value>
 </property>
 <property>
    <name>hbase.zookeeper.quorum</name>
    <value>localhost:2181</value>
 </property>
 <property>
    <name>zookeeper.znode.parent</name>
    <value>/hbase</value>
 </property>
</configuration>

For this example, we will create a test table in HBase. If you already have a table in HBase, feel free to use that table instead.

Launch the HBase shell and create a table named fusion_nums with a single column family named lw:
create 'fusion_nums', 'lw'

Do a list command to see your table:

hbase(main):002:0> list
TABLE
fusion_nums
    1 row(s) in 0.0250 seconds

=> ["fusion_nums"]

Fill the table with some data:

for i in '1'..'100' do for j in '1'..'2' do put 'fusion_nums', "row#{i}", "lw:c#{j}", "#{i}#{j}" end end

Scan the fusion_nums table to see your data:
scan 'fusion_nums'

The HBase connector requires a catalog read option that defines the columns you want to read and how to map them into a Spark DataFrame. For our sample table, the following suffices:

{
    "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"}
    }
 }

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:

{
  "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:

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:

{
  "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.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.

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.
Connect to Couchbase using the command line client cbq. For example, cbq -e=http://<host>:8091 -u <user> -p <password>. Ensure the provided user is an authorized user of the test bucket.
Create a primary index on the test bucket: CREATE PRIMARY INDEX 'test-primary-index' ON 'test' USING GSI;.
Insert some data: INSERT INTO 'test' ( KEY, VALUE ) VALUES ( "1", { "id": "01", "field1": "a value", "field2": "another value"} ) RETURNING META().id as docid, *;.
Verify you can query the document just inserted: select * from 'test';.

To ingest from this bucket with the Parallel Bulk Loader, use the Couchbase Spark connector by specifying the format com.couchbase.spark.sql.DefaultSource. Then specify the com.couchbase.client:spark-connector_2.11:2.2.0 package as the spark shell --packages option, as well as a few spark settings that direct the connector to a particular Couchbase server and bucket to connect to using the provided credentials. See here for all of the available Spark configuration settings for the Couchbase Spark connector.Putting it all together:

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:

{
    "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
}

API usage examples

The following examples include requests and responses for some of the API endpoints.

Create a parallel bulk loader job

An example request to create a parallel bulk loader job with the REST API is as follows:

curl --request POST \
  --url https://FUSION_HOST/api/spark/configurations \
  --header "Accept: */*" \
  --header "Authorization: Basic ACCESS_TOKEN" \
  --header "Content-Type: application/json" \
  --data '{
  "id": "store_typeahead_entity_load",
  "format": "solr",
  "sparkConfig": [
    {
      "key": "spark.sql.caseSensitive",
      "value": "true"
    },
  ],
  "readOptions": [
        {
      "key": "collection",
      "value": "store"
    },
    {
      "key": "zkHost",
      "value": "zookeeper-549"
    },
  ],
  "type": "parallel-bulk-loader"
}

The response is a message indicating success or failure.

Start a parallel bulk loader job

This request starts the pbl_load PBL job.

curl --request POST \
  --url https://FUSION_HOST.com/api/spark/jobs/store_typeahead_entity_load \
  --header "Accept: */*" \
  --header "Authorization: Basic ACCESS_TOKEN" \
  --header "Content-Type: application/json"

The response is:

{
    "state": "starting",
    "jobId": "a2b50rrce7",
    "jobConfig": {
        "type": "parallel-bulk-loader",
        "id": "store_typeahead_entity_load",
        "format": "solr",
        "readOptions": [
            {
                "key": "collection",
                "value": "store"
            },
            {
                "key": "zkHost",
                "value": "zookeeper-549"
            }
        ],
        "outputCollection": "store_typeahead",
        "outputIndexPipeline": "store_typeahead",
        "clearDatasource": true,
        "defineFieldsUsingInputSchema": true,
        "atomicUpdates": false,
        "transformScala": "import script",
        "sparkConfig": [
            {
                "key": "spark.sql.caseSensitive",
                "value": "true"
            },
            {
                "key": "spark.typeField_1",
                "value": "Suggested_Doc, doc_title, url, doc_locale, doc_sector_ss, doc_economic_buyer_ss, doc_topic_ss, title_s"
            }
        ],
        "cacheAfterRead": false,
        "continueAfterFailure": false,
        "type": "parallel-bulk-loader",
        "updates": [
            {
                "userId": "service_account",
                "timestamp": "2024-05-06T09:06:43.739877Z"
            },
            {
                "userId": "power_user",
                "timestamp": "2024-07-30T20:30:31.347930292Z"
            },
            {
                "userId": "power_user",
                "timestamp": "2024-07-30T20:30:31.350243642Z"
            }
        ]
    },
    "hostname": "111.111.111.111",
    "result": {
        "jobConfigId": "store_typeahead_entity_load",
        "jobRunId": "b2c50rrde6"
    }
}

POST spark configuration to a parallel bulk loader job

This request posts Spark configuration information to the store_typeahead_entity_load PBL job.

curl --request POST \
  --url https://FUSION_HOST.com/api/spark/configurations \
  --header "Accept: */*" \
  --header "Authorization: Basic ACCESS_TOKEN" \
  --header "Content-Type: application/json" \
  --data '{
  "id": "store_typeahead_entity_load",
  "sparkConfig": [
    {
      "key": "spark.sql.caseSensitive",
      "value": "true",
    },
    {
      "key": "spark.typeField_1",
      "value": "Suggested_Doc, doc_title, url, doc_locale, doc_sector_ss, doc_economic_buyer_ss, doc_topic_ss, title_s",
    }
  ],
  "type": "parallel-bulk-loader"
}'

Learn more

Import Solr Collections with the Parallel Bulk Loader

You can use the Parallel Bulk Loader to copy Solr collections from one collection to another. This is helpful for copying collections from a production environment to a testing or development environment and using real data in the development and testing process.

Parallel Bulk Loader Job Configuration

In the Lucidworks Search UI:

Navigate to Collections > Jobs.
Click Add and select Parallel Bulk Loader from the menu.
Enter a name for your job, and enter solr as the format.
Set the parameter name and value for the collection you want to import.
- Parameter name: collection
- Parameter value: the name of the collection to import from
If the source collection is in a different Lucidworks Search app, add an additional parameter name and value pair.
- Parameter name: zkHost
- Parameter value: the location of zkHost
Enter the output collection, or where you want to import the collection to.
- Output collection: the name of the output collection
- Send to index pipeline: _system
Save your job.

When you’re ready, run the job or schedule the job to run at an interval of your choice.

Parallel Bulk Loader JSON Configuration

Use this sample JSON payload to load the Parallel Bulk Loader to the Spark Jobs API:

{
  "type" : "parallel-bulk-loader",
  "id" : "copy_sigs_with_timestamp",
  "format" : "solr",
  "readOptions" : [ {
    "key" : "collection",
    "value" : "SOURCE_COLLECTION"
  }, {
    "key" : "query",
    "value" : "timestamp_tdt:{$lastTimestamp*}* TO *]"
  } ],
  "outputCollection" : "OUTPUT_COLLECTION",
  "timestampFieldName" : "timestamp_tdt",
  "defineFieldsUsingInputSchema" : true
}

Replace SOURCE_COLLECTION with the name of the collection being exported. Replace OUTPUT_COLLECTION with the name of the destination collection.Use the following Spark Jobs API call to run the job. Replace data.json with your JSON payload file.

curl -X POST \
-u USERNAME:PASSWORD \
'https://FUSION_HOST:6764/api/apps/Documentation/spark/configurations' \
-H 'accept: */*' \
-H 'Content-Type: application/json' \
-d 'data.json'

Import with the Bulk Loader

Create and run Parallel Bulk Loader jobs

Configuration settings for the Parallel Bulk Loader job

This section provides configuration settings for the Parallel Bulk Loader job.

Read settings

Setting	Description
`format`	Unique identifier of the data source provider. Spark scans the job’s `classpath` for a class named `DefaultSource` in the `<format>` package. For example, for the `solr` format, we provide the `solr.DefaultSource` class in our `spark-solr` repository
`path` (optional)	Comma-delimited list of paths to load. Some data sources, such as parquet, require a path. Others, such as Solr, do not. Refer to the documentation for your data source to determine if you need to provide a path.
`readOptions`	Options passed to the Spark SQL data source to configure the read operation. Options differ for every data source. Refer to the specific data source documentation for more information.
`sparkConfig` (optional)	List of Spark configuration settings needed to run the Parallel Bulk Loader.
`shellOptions`	Behind the scenes, the Parallel Bulk Loader job submits a Scala script to the Fusion Spark shell. The `shellOptions` setting lets you pass any additional options needed by the Spark shell. The two most common options are `--packages` and `--repositories`: `--packages` Comma-separated list of Maven coordinates of JAR files to include on the driver and executor classpaths. Spark searches the local Maven repository, and then Maven central and any additional remote repositories given in the config. The format for the coordinates should be `groupId:artifactId:version`. The HBase example below demonstrates the use of the packages option for loading the `com.hortonworks:shc-core:1.1.1-2.1-s_2.11` package. Use the `https://spark-packages.org/` site to find interesting packages to add to your Parallel Bulk Loader jobs. `--repositories` Comma-separated list of additional remote Maven repositories to search for the Maven coordinates given in the `packages` config setting. The Index HBase tables example below demonstrates the use of the `repositories` option for loading the `com.hortonworks:shc-core:1.1.1-2.1-s_2.11` package from this repository.
`timestampFieldName`	For datasources that support time-based filters, the Parallel Bulk Loader computes the timestamp of the last document written to Solr and the current timestamp of the Parallel Bulk Loader job. For example, the HBase data source lets you filter the read between a `MIN_STAMP` and `MAX_STAMP`, for example: `val timeRangeOpts = Map(HBaseRelation.MIN_STAMP -> minStamp.toString, HBaseRelation.MAX_STAMP -> maxStamp.toString)` lets Parallel Bulk Loader jobs run on schedules, and pull only the newest rows from the underlying datasources. To support timestamp based filtering, the Parallel Bulk Loader provides two simple macros: `$lastTimestamp(format)` `$nowTimestamp(format)` The `format` argument is optional. If not supplied, then an ISO-8601 date/time string is used. The `timestampFieldName` setting is used to determine the value of `lastTimestamp`, using a Top 1 query to Solr to get the max timestamp. You can also pass `$lastTimestamp(EPOCH)` or `$lastTimestamp(EPOCH_MS)` to get the timestamp in seconds or milliseconds. See the Index HBase tables example below for an example of using this configuration property.

Transformation settings

Setting	Description
`transformScala`	Sometimes, you can write a small script to transform input data into the correct form for indexing. But at other times, you might need the full power of the Spark API to transform data into an indexable form. The `transformScala` option lets you filter and/or transform the input DataFrame any way you would like. You can even define UDFs to use during your transformation. For an example of using Scala to transform the input DataFrame before indexing in Solr, see the Read from Parquet example. Another powerful use of the `transformScala` option is that you can pull in advanced libraries, such as Spark NLP (from John Snow Labs) to do NLP work on your content before indexing. See the Use NLP during indexing example. Your Scala script can do other things but, at a minimum, it must define the function that the Parallel Bulk Loader invokes (see function below this table.)
`transformSql`	The `transformSql` option lets you write a SQL query to transform the input DataFrame. The SQL is executed after the `transformScala` script (if both are defined). The input DataFrame is exposed to your SQL as the `_input` view. See the Clean up data with SQL transformations example below for an example of using SQL to transform the input before indexing in Solr. This option also lets you leverage the UDF/UDAF functions provided by Spark SQL.
`mlModelId`	If you have a Spark ML PipelineModel loaded into the blob store, you can supply the blob ID to the Parallel Bulk Loader and it will: Load the model from the blob store. Transform the input DataFrame (after the Scala transform but before the SQL transform). 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 Parallel Bulk Loader:

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:<JOB>`. Consequently, the Parallel Bulk Loader adds two metadata fields to each row: `_lw_loader_id_s` and `_lw_loader_job_id_s`.
`atomicUpdates`	Flag to send documents directly to Solr as atomic updates instead of as new documents. This option is not supported when using an index profile. Also note that the Parallel Bulk Loader tracking fields `_lw_loader_id_s` and `_lw_loader_job_id_s` are not sent when using atomic updates, so the clear datasource option does not work with documents created using atomic updates.
`outputOptions`	Options used when writing directly to Solr. See Spark-Solr: https://github.com/lucidworks/spark-solr#index-parameters For example, if your docs are relatively small, you might want to increase the `batch_size` (2000 default) as shown below:
`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.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:

Use this transform Scala script:

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

Use the following SQL to clean up the input data before indexing:

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

<configuration>
  <property>
    <name>fs.s3a.access.key</name>
    <value>???</value>
  </property>
  <property>
    <name>fs.s3a.secret.key</name>
    <value>???</value>
  </property>
</configuration>

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.

Read from Parquet

Reading from parquet files is built into Spark using the “parquet” format. For additional read options, see Configuration of Parquet.

Job JSON:

{
  "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:

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

Load the JDBC driver JAR file into the Blob store

curl -XPUT -H "Content-type:application/octet-stream" "http://<FUSION_HOST>/api/blobs/mysql_jdbc_jar?resourceType=spark:jar" --data-binary @mysql-connector-java-5.1.45-bin.jar

Read from a table

For more information on reading from JDBC-compliant databases, see:

{
  "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
}

Index HBase tables

To index an HBase table, use the Hortonworks connector.

The Parallel Bulk Loader lets us replace the HBase Indexer.

You will need to add an hbase-site.xml (and possibly core-site.xml) to apps/spark-dist/conf in Fusion, for example:

<configuration>
 <property>
    <name>hbase.defaults.for.version.skip</name>
    <value>true</value>
 </property>
 <property>
    <name>hbase.zookeeper.quorum</name>
    <value>localhost:2181</value>
 </property>
 <property>
    <name>zookeeper.znode.parent</name>
    <value>/hbase</value>
 </property>
</configuration>

For this example, we will create a test table in HBase. If you already have a table in HBase, feel free to use that table instead.

Launch the HBase shell and create a table named fusion_nums with a single column family named lw:
create 'fusion_nums', 'lw'

Do a list command to see your table:

hbase(main):002:0> list
TABLE
fusion_nums
1 row(s) in 0.0250 seconds
=> ["fusion_nums"]

Fill the table with some data:

for i in '1'..'100' do for j in '1'..'2' do put 'fusion_nums', "row#{i}", "lw:c#{j}", "#{i}#{j}" end end

Scan the fusion_nums table to see your data:
scan 'fusion_nums'

The HBase connector requires a catalog read option that defines the columns you want to read and how to map them into a Spark DataFrame. For our sample table, the following suffices:

{
    "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"}
    }
 }

{
  "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:

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:

{
  "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

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.
Connect to Couchbase using the command line client cbq. For example, cbq -e=http://<host>:8091 -u <user> -p <password>. Ensure the provided user is an authorized user of the test bucket.
Create a primary index on the test bucket: CREATE PRIMARY INDEX 'test-primary-index' ON 'test' USING GSI;.
Insert some data: INSERT INTO 'test' ( KEY, VALUE ) VALUES ( "1", { "id": "01", "field1": "a value", "field2": "another value"} ) RETURNING META().id as docid, *;.
Verify you can query the document just inserted: select * from 'test';.

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:

{
    "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
}

​Usage

​Example configuration

​Additional field information

​Create and run Parallel Bulk Loader jobs

​Configuration settings for the Parallel Bulk Loader job

​Read settings

​Transformation settings

​Output settings

​Tune performance

​Examples

​Use NLP during indexing

​Clean up data with SQL transformations

​Read from S3

​Read from Parquet

​Read from JDBC tables

​Load the JDBC driver JAR file into the Blob store

​Read from a table

​Index HBase tables

​Index Elastic data

​Read from Couchbase

​XML setup

​API usage examples

​Create a parallel bulk loader job

​Start a parallel bulk loader job

​POST spark configuration to a parallel bulk loader job

​Learn more

​Parallel Bulk Loader Job Configuration

​Parallel Bulk Loader JSON Configuration

​Create and run Parallel Bulk Loader jobs

​Configuration settings for the Parallel Bulk Loader job

​Read settings

​Transformation settings

​Output settings

​Tune performance

​Examples

​Use NLP during indexing

​Clean up data with SQL transformations

​Read from S3

​Read from Parquet

​Read from JDBC tables

​Load the JDBC driver JAR file into the Blob store

​Read from a table

​Index HBase tables

​Index Elastic data

​Read from Couchbase

​XML setup

​Configuration properties

Usage

Example configuration

Additional field information

Create and run Parallel Bulk Loader jobs

Configuration settings for the Parallel Bulk Loader job

Read settings

Transformation settings

Output settings

Tune performance

Examples

Use NLP during indexing

Clean up data with SQL transformations

Read from S3

Read from Parquet

Read from JDBC tables

Load the JDBC driver JAR file into the Blob store

Read from a table

Index HBase tables

Index Elastic data

Read from Couchbase

XML setup

API usage examples

Create a parallel bulk loader job

Start a parallel bulk loader job

POST spark configuration to a parallel bulk loader job

Learn more

Parallel Bulk Loader Job Configuration

Parallel Bulk Loader JSON Configuration

Create and run Parallel Bulk Loader jobs

Configuration settings for the Parallel Bulk Loader job

Read settings

Transformation settings

Output settings

Tune performance

Examples

Use NLP during indexing

Clean up data with SQL transformations

Read from S3

Read from Parquet

Read from JDBC tables

Load the JDBC driver JAR file into the Blob store

Read from a table

Index HBase tables

Index Elastic data

Read from Couchbase

XML setup

Configuration properties