PySpark

The PySpark integration relies on two components:

The Python library is a prerequisite to include the PySpark agent. Please make sure to have installed the Python Kensu-py library.

To install the library, please execute this command in your Python environment:

Download a jar to match your Spark version:

Ideally it should be a full verison match, but often a major.minor match is good enough too (i.e. last version number may differ).

To include the PySpark agent inside, you need to:

Here is a summary of the code modification you need to perform to include pyspark agent:

The Spark agent can also be included during the spark-submit command execution:

Here the conf.ini template:

You will find below this page the list of custom parameters you can add to this configuration file.

The basic parameters of the configuration file are the following:

Metrics computation in Kensu PySpark agent can be controlled by using these parameters in the conf.ini file as described below. Metrics computation can also be fine-tuned for each Logical Datasource usage & Application in the Kensu web app.

In Kensu web app, by going to Logical Datasource summary page, you may choose which statistics for which columns to compute or disable them for each individual datasource access in a given application.

Remote conf takes priority over conf.ini

By default, the remote configuration is enabled, and if configured on the Kensu app, it would override the behaviour specified in conf.inibut only for that specific Application & Datasource.

How to disable remote conf?

If for any reason, you want to use only the conf.ini and disable usage of the remote configuration (e.g. for testing new global settings), set remote_conf_enabled=False parameter.

How to debug, if needed?

Actual metrics configuration used for each datasource is printed in Kensu logs during metrics computation, indicating which configuration source was used for each datasource

Add the following code, at your PySpark at places where you want the circuit breaking to happen.

If not already done, define some rules in Kensu app, and mark some of them as circuit-breaker for the given Spark app.

The circuit breaker will kill the Spark application if any of the datasources that were already accessed (i.e. read or written) by Spark has any open tickets due to broken circuit breaker rules in Kensu at Logical datasource level for any of the previous runs (only if the associated circuit breaker rule is activated for the current Spark app).

If you have more complex use-case, you may also perform your own handling by just checking the circuit breaker status. Don't forget to properly shutdown spark & kensu spark agent if you need to quit the application.

Kensu tracks every data source based on its location (i.e. full path of a file dbfs:/FileStore/Financial_Data_Report/2021/dec/AppTech.csv), and there are different option to choose how to name the data source. In the following example only the last folder and filename is used:

In addition, Kensu introduces also the concept of "logical" data sources, this can be useful to group data sources (that are actually partitions) to be part of the same logical data source. As an example of logical data source usage, consider to have these physical paths:

In this case these 2 data sources can be grouped into dbfs://application/OUTPUT/Customers , this can be achieved with the available logical_data_source_naming_strategy options:

Assuming the original filepath is dbfs:/FileStore/Financial_Data_Report/2021/dec/AppTech.csv, these are the results of the different options:

The data_source_naming_strategy_rules allows to set multiple rules (separated by ;). The rules could be path specific (i.e. dbfs->>LastFolderAndFile) or using regular expression in the following format: regexp-match:<regular expression>->>replace:<replace expression>

Consider to have in one application the following data sources:

Applying the following rules:

data_source_naming_strategy_rules=regexp-match:Spark Catalog: default.(.*)->>replace:database:$1;dbfs->>LastTwoFoldersAndFile

This is the result:

Because the first rule is a regular expression with a substitution and the second one is a default path search with a standard strategy option. The first expression that matches is taken into account.

If no expression matches, the default strategy is File.

Similarly for logical_data_source_naming_strategy_rules if we have two data sources:

to obtain that these 2 data sources are grouped under dbfs://application/OUTPUT/Customers this configuration should be added:

logical_data_source_naming_strategy_rules=regexp-match:(.*)/\\d*/(.*)->>replace:$1/$2

The PySpark agent sends metadata in batches to Kensu each time a write operation of a data source is finished successfully, whatever the format.

e.g.: spark.write.save() operations

P.S. we currently do not report anything when such write operation that did not happen due to say error in code or a Spark/cluster error, this is so that records in Kensu match the reallity

The Kensu PySpark agent relies on the schema of the data source in order to define the set of automatic metrics it has to retrieve. To make sure the data type is well inferred, you need to add the inferschema option while reading some schema-less data source types like CSV.

e.g. df = spark.read.option("inferschema","true").csv("/path/to/ds")

Whether the logs are written in the log file, depends on the log level of your spark session. A lot of issues with the log file can be troubleshot by adding the following line before calling the init_kensu_spark() function:

2. Create a Configuration file

3. Initialize the Kensu agent in the your Spark job script