Avro files

For using Avro files in Spark mappings, the Avro .egg file has to be added to the ODI installation. The .egg file for Avro cannot be downloaded directly, and has to be generated from the Avro package.

To add the Avro .egg file to the ODI installation:

Delimited files

For using Delimited files in Spark mappings, external jar files must be added to the ODI installation.

To add the CSV jar files to the ODI installation:

Resilient Distributed Datasets (RDD)

RDDs are the primary data abstraction in Apache Spark. They are fault tolerant (Resilient) collections of partitioned data (Dataset) with data residing on multiple nodes in a cluster (Distributed). The data resides in memory and is cached where necessary. RDDs are read-only, but can have transformations applied that will create other RDDs. Lazy evaluation is used, where the data is only available or transformed when triggered. RDD partitions are the unit of parallelism.

DataFrames

A DataFrame is a read-only distributed collection of data, that (unlike RDDs) is organized into named columns. The abstraction level is higher, making the processing of large datasets even easier, such as in allowing the use of SparkSQL queries. DataFrames are built on top of the Spark SQL engine, allowing for much better performance and space optimization.

Spark can infer or deduce the Schema of a dataset by looking at the data. Although this can be advantageous, there are some circumstances where datatypes may not be mapped as expected. If this happens, there is an inferSchema option on applicable Spark KMs that can be set to False, turning off this functionality. If you see runtime errors related to datatype mismatches, you may need to adjust the value of the Infer Schema option. This option can be set on reading or writing LKMs.

The Infer Schema option can be seen in the figure below.

Figure 7-1 Physical Mapping with InferSchema KM Option

Description of "Figure 7-1 Physical Mapping with InferSchema KM Option"

When you need to write expressions, for example, in a Join or Filter condition, or an Attribute expression, you have options that can be used for the Expression Syntax. If you have decided to have ODI generate RDD code, then your expressions must be written in Python. If, however, you have decided to generate DataFrames, then you can choose to write your expressions in SQL or Python. You can specify your chosen syntax by setting SQL_EXPRESSIONS to True/False.

The combinations of possible code generation style are:

Since Python expressions are defined differently in RDD and DataFrames, the Python syntax for these two styles of code generation can be different. Therefore, not all Python expressions will work for both RDD and DataFrame code generation styles.

RDD with Python expressions

For information on the syntax and functions that can be used in Python expressions, see The Python Standard Library.

DataFrames with Python expressions

For information on the list of Python functions available to Column objects, see Pyspark.sql.module.

DataFrames with SQL expressions

The generic SQL functions and operators can be viewed in the Expression editor on selecting generic SQL language.

Mapping Description

Consider an example that shows multiple expressions being used in mappings.

In this example, a source (REA) containing Real Estate Transactions is combined with a second source (REA2) containing City and Population data. A filter is then applied to select only large transactions. This creates a target file (REA1) which contains the combined and filtered information as shown in the figure below.

Figure 7-2 Mapping with Multiple Expressions

Description of "Figure 7-2 Mapping with Multiple Expressions"

Mapping Definitions

The mapping is defined as follows:

Mapping Expressions for each Codegen Style

The following table describes the mapping expressions for each codegen style.

REA.City == (REA2.City).upper()

REA.City == upper(REA2.City)

REA.City = UPPER(REA2.City)

REA.Type<>'Multi-Family' and REA.Price >=500000

REA.Type<>'Multi-Family' and  REA.Price >=500000

REA.Type<>'Multi-Family' and  REA.Price >=500000

# City - note: this only capitalizes the first word!
(REA.City).capitalize()
# GeoLocation
REA.geolat + ", " +  REA.geolong
# Population
int(round(REA2.Population,-3))

# City
initcap(lower(REA.City))
# GeoLocation
concat(REA.geolat ,lit(", "),REA.geolong)
# Population
round(REA2.Population,-3)

-- City
INITCAP(LOWER(REA.City))
-- GeoLocation
CONCAT(REA.geolat,', ', REA.geolong)
# Population
ROUND(REA2.Population,-3)

Importing Libraries

As you'll see from this example, not all the same built-in functions are available across these differing styles. In this case, the initcap built-in function is not available in RDD. The capwords() function does what is required, but it requires import statements to be added to the script. The Spark EKM has a multi line option called customPythonImports that lets you specify the Import Statements for the script, thereby allowing extra functions to be available in the expressions.

To contain the list of imports, the customPythonImports EKM option will be written as

from string import *
from time import localtime

The Target Column expression would then be written as

#City
capwords(REA.City)

Stateful Aggregation

When data must be aggregated across all data of a stream, stateful aggregation is required. In stateful aggregation Spark builds called state stream containing the aggregated values for all keys. For every incoming RDD this state is updated, for example aggregated sums are updated based on new incoming data.

By default a state stream will output all stored values for every incoming RDD. This is useful in case the stream output is a file and the file is expected to always hold the entire set of aggregate values.

Stateful processing is supported by:

Caching

In ODI, the Spark caching mechanism is leveraged by providing two additional Spark base KM options.

• Cache data: If this option set to true a storage invocation is added to the generated pyspark code of the component.

• Cache storage level: This option is hidden if cache data is set to false.

Repartitioning

If the source is a HDFS file, the number of partitions is initially determined by the data block of the source HDFS system. The platform resource is not fully used if the platform that runs the Spark application has more available slots for running tasks than the number of partitions loaded. In such cases, the RDD.repartition() api can be used to change the number of partitions.

Repartitioning can be done in any step of the whole process, even immediately after data is loaded from source or after processing the filter component. ODI has Spark base KM options which let you decide whether and where to do repartitioning.

• Repartition: If this option is set to true, repartition is applied after the transformation of component.

• Level of Parallelism: Number of partitions and the default is 0. When the default value is set, spark.default.parallelism will be used to invoke the repartition() function.

• Sort Partitions: If this option is set to true, partitions are sorted by key and the key is defined by a Lambda function.

• Partitions Sort Order: Ascending or descending. Default is ascending.

• Partition Keys: User defined partition keys represented as a comma separated column list.

• Partition Function: User defined partition Lambda function. Default value is a pyspark defined hash function portable_hash, which simply computes a hash base on the entire RDD row.