How To Log Databricks Workflows with the Elastic (ELK) Stack | by Yury Kalbaska | Jul, 2024

A sensible instance of establishing observability for an information pipeline utilizing finest practices from SWE world

On the time of this writing (July 2024) Databricks has turn into an ordinary platform for knowledge engineering within the cloud, this rise to prominence highlights the significance of options that assist strong knowledge operations (DataOps). Amongst these options, observability capabilities — logging, monitoring, and alerting — are important for a mature and production-ready knowledge engineering software.

There are numerous instruments to log, monitor, and alert the Databricks workflows together with built-in native Databricks Dashboards, Azure Monitor, DataDog amongst others.

Nevertheless, one widespread state of affairs that’s not clearly coated by the above is the necessity to combine with an present enterprise monitoring and alerting stack slightly than utilizing the devoted instruments talked about above. As a rule, this can be Elastic stack (aka ELK) — a de-facto normal for logging and monitoring within the software program growth world.

Parts of the ELK stack?

ELK stands for Elasticsearch, Logstash, and Kibana — three merchandise from Elastic that supply end-to-end observability answer:

1. Elasticsearch — for log storage and retrieval
2. Logstash — for log ingestion
3. Kibana — for visualizations and alerting

The next sections will current a sensible instance of methods to combine the ELK Stack with Databricks to attain a sturdy end-to-end observability answer.

Stipulations

Earlier than we transfer on to implementation, guarantee the next is in place:

1. Elastic cluster — A operating Elastic cluster is required. For easier use instances, this generally is a single-node setup. Nevertheless, one of many key benefits of the ELK is that it’s absolutely distributed so in a bigger group you’ll in all probability cope with a cluster operating in Kubernetes. Alternatively, an occasion of Elastic Cloud can be utilized, which is equal for the needs of this instance.
   If you’re experimenting, seek advice from the excellent guide by DigitalOcean on methods to deploy an Elastic cluster to a neighborhood (or cloud) VM.
2. Databricks workspace — guarantee you’ve permissions to configure cluster-scoped init scripts. Administrator rights are required in the event you intend to arrange world init scripts.

Storage

For log storage, we’ll use Elasticsearch’s personal storage capabilities. We begin by establishing. In Elasticsearch knowledge is organized in indices. Every index accommodates a number of paperwork, that are JSON-formatted knowledge buildings. Earlier than storing logs, an index should be created. This process is typically dealt with by a company’s infrastructure or operations group, but when not, it may be achieved with the next command:

curl -X PUT "http://localhost:9200/logs_index?fairly"

Additional customization of the index might be achieved as wanted. For detailed configuration choices, seek advice from the REST API Reference: https://www.elastic.co/guide/en/elasticsearch/reference/current/indices-create-index.html

As soon as the index is about up paperwork might be added with:

curl -X POST "http://localhost:9200/logs_index/_doc?fairly"
-H 'Content material-Kind: utility/json'
-d'
{
"timestamp": "2024-07-21T12:00:00",
"log_level": "INFO",
"message": "This can be a log message."
}'

To retrieve paperwork, use:

curl -X GET "http://localhost:9200/logs_index/_search?fairly"
-H 'Content material-Kind: utility/json'
-d'
{
"question": {
"match": {
"message": "This can be a log message."
}
}
}'

This covers the important performance of Elasticsearch for our functions. Subsequent, we’ll arrange the log ingestion course of.

Transport / Ingestion

Within the ELK stack, Logstash is the element that’s chargeable for ingesting logs into Elasticsearch.

The performance of Logstash is organized into pipelines, which handle the stream of knowledge from ingestion to output.

Every pipeline can encompass three primary phases:

1. Enter: Logstash can ingest knowledge from varied sources. On this instance, we’ll use Filebeat, a light-weight shipper, as our enter supply to gather and ahead log knowledge — extra on this later.
2. Filter: This stage processes the incoming knowledge. Whereas Logstash helps varied filters for parsing and reworking logs, we won’t be implementing any filters on this state of affairs.
3. Output: The ultimate stage sends the processed knowledge to a number of locations. Right here, the output vacation spot can be an Elasticsearch cluster.

Pipeline configurations are outlined in YAML recordsdata and saved within the /and so forth/logstash/conf.d/ listing. Upon beginning the Logstash service, these configuration recordsdata are mechanically loaded and executed.

You’ll be able to seek advice from Logstash documentation on methods to arrange one. An instance of a minimal pipeline configuration is offered under:

enter {
beats {
port => 5044
}
}
filter {}
output {
elasticsearch {
hosts => ["http://localhost:9200"]
index => "filebeat-logs-%{+YYYY.MM.dd}"
}
}

Lastly, make sure the configuration is right:

bin/logstash -f /and so forth/logstash/conf.d/take a look at.conf --config.test_and_exit

Amassing utility logs

There’s yet another element in ELK — Beats. Beats are light-weight brokers (shippers) which can be used to ship log (and different) knowledge into both Logstash or Elasticsearch immediately. There’s a lot of Beats — every for its particular person use case however we’ll consider Filebeat — by far the most well-liked one — which is used to gather log recordsdata, course of them, and push to Logstash or Elasticsearch immediately.

Beats should be put in on the machines the place logs are generated. In Databricks we’ll must setup Filebeat on each cluster that we need to log from — both All-Goal (for prototyping, debugging in notebooks and comparable) or Job (for precise workloads). Putting in Filebeat entails three steps:

1. Set up itself — obtain and execute distributable package deal on your working system (Databricks clusters are operating Ubuntu — so a Debian package deal needs to be used)
2. Configure the put in occasion
3. Beginning the service by way of system.d and asserting it’s lively standing

This may be achieved with the assistance of Init scripts. A minimal instance Init script is usually recommended under:

#!/bin/bash
# Test if the script is run as root
if [ "$EUID" -ne 0 ]; then
echo "Please run as root"
exit 1
fi
# Obtain filebeat set up package deal
SRC_URL="https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-8.14.3-amd64.deb"
DEST_DIR="/tmp"
FILENAME=$(basename "$SRC_URL")
wget -q -O "$DEST_DIR/$FILENAME" "$SRC_URL"
# Set up filebeat
export DEBIAN_FRONTEND=noninteractive
dpkg -i /tmp/filebeat-8.14.3-amd64.deb
apt-get -f set up -y
# Configure filebeat
cp /and so forth/filebeat/filebeat.yml /and so forth/filebeat/filebeat_backup.yml
tee /and so forth/filebeat/filebeat.yml > /dev/null <<EOL
filebeat.inputs:
- kind: filestream
id: my-application-filestream-001
enabled: true
paths:
- /var/log/myapplication/*.txt
parsers:
- ndjson:
keys_under_root: true
overwrite_keys: true
add_error_key: true
expand_keys: true
processors:
- timestamp:
subject: timestamp
layouts:
- "2006-01-02T15:04:05Z"
- "2006-01-02T15:04:05.0Z"
- "2006-01-02T15:04:05.00Z"
- "2006-01-02T15:04:05.000Z"
- "2006-01-02T15:04:05.0000Z"
- "2006-01-02T15:04:05.00000Z"
- "2006-01-02T15:04:05.000000Z"
take a look at:
- "2024-07-19T09:45:20.754Z"
- "2024-07-19T09:40:26.701Z"
output.logstash:
hosts: ["localhost:5044"]
logging:
stage: debug
to_files: true
recordsdata:
path: /var/log/filebeat
title: filebeat
keepfiles: 7
permissions: 0644
EOL
# Begin filebeat service
systemctl begin filebeat
# Confirm standing
# systemctl standing filebeat

Timestamp Difficulty

Discover how within the configuration above we arrange a processor to extract timestamps. That is achieved to deal with a typical drawback with Filebeat — by default it’s going to populate logs @timestamp subject with a timestamp when logs had been harvested from the designated listing — not with the timestamp of the particular occasion. Though the distinction isn’t greater than 2–3 seconds for lots of purposes, this may mess up the logs actual dangerous — extra particularly, it may possibly mess up the order of data as they’re coming in.

To handle this, we’ll overwrite the default @timestamp subject with values from log themselves.

Logging

As soon as Filebeat is put in and operating, it’s going to mechanically accumulate all logs output to the designated listing, forwarding them to Logstash and subsequently down the pipeline.

Earlier than this may happen, we have to configure the Python logging library.

The primary crucial modification could be to arrange FileHandler to output logs as recordsdata to the designated listing. Default logging FileHandler will work simply high-quality.

Then we have to format the logs into NDJSON, which is required for correct parsing by Filebeat. Since this format isn’t natively supported by the usual Python library, we might want to implement a customized Formatter.

class NDJSONFormatter(logging.Formatter):
def __init__(self, extra_fields=None):
tremendous().__init__()
self.extra_fields = extra_fields if extra_fields isn't None else {}
def format(self, document):
log_record = {
"timestamp": datetime.datetime.fromtimestamp(document.created).isoformat() + 'Z',
"log.stage": document.levelname.decrease(),
"message": document.getMessage(),
"logger.title": document.title,
"path": document.pathname,
"lineno": document.lineno,
"operate": document.funcName,
"pid": document.course of,
}
log_record = {**log_record, **self.extra_fields}
if document.exc_info:
log_record["exception"] = self.formatException(document.exc_info)
return json.dumps(log_record)

We may even use the customized Formatter to deal with the timestamp challenge we mentioned earlier. Within the configuration above a brand new subject timestamp is added to the LogRecord object that can conatain a duplicate of the occasion timestamp. This subject could also be utilized in timestamp processor in Filebeat to interchange the precise @timestamp subject within the printed logs.

We are able to additionally use the Formatter so as to add further fields — which can be helpful for distinguishing logs in case your group makes use of one index to gather logs from a number of purposes.

Further modifications might be made as per your necessities. As soon as the Logger has been arrange we will use the usual Python logging API — .data() and .debug(), to put in writing logs to the log file and they’ll mechanically propagate to Filebeat, then to Logstash, then to Elasticsearch and eventually we can entry these in Kibana (or another shopper of our alternative).

Visualization

Within the ELK stack, Kibana is a element chargeable for visualizing the logs (or another). For the aim of this instance, we’ll simply use it as a glorified search shopper for Elasticsearch. It could actually nevertheless (and is meant to) be arrange as a full-featured monitoring and alerting answer given its wealthy knowledge presentation toolset.

So as to lastly see our log knowledge in Kibana, we have to arrange Index Patterns:

1. Navigate to Kibana.
2. Open the “Burger Menu” (≡).
3. Go to Administration -> Stack Administration -> Kibana -> Index Patterns.
4. Click on on Create Index Sample.

Kibana will helpfully recommend names of the obtainable sources for the Index Patterns. Kind out a reputation that can seize the names of the sources. On this instance it may be e.g. filebeat*, then click on Create index sample.

As soon as chosen, proceed to Uncover menu, choose the newly created index sample on the left drop-down menu, regulate time interval (a typical pitfall — it’s set as much as final quarter-hour by default) and begin with your personal first KQL question to retrieve the logs.

Now we have now efficiently accomplished the multi-step journey from producing a log entry in a Python utility hosted on Databricks to to visualizing and monitoring this knowledge utilizing a shopper interface.

Whereas this text has coated the introductory points of establishing a sturdy logging and monitoring answer utilizing the ELK Stack along side Databricks, there are extra issues and superior matters that recommend additional exploration:

• Selecting Between Logstash and Direct Ingestion: Evaluating whether or not to make use of Logstash for extra knowledge processing capabilities versus immediately forwarding logs from Filebeat to Elasticsearch.
• Schema Concerns: Deciding on the adoption of the Elastic Widespread Schema (ECS) versus implementing customized subject buildings for log knowledge.
• Exploring Various Options: Investigating different instruments equivalent to Azure EventHubs and different potential log shippers that will higher match particular use instances.
• Broadening the Scope: Extending these practices to embody different knowledge engineering instruments and platforms, making certain complete observability throughout the complete knowledge pipeline.

These matters can be explored in additional articles.

Until in any other case famous, all photographs are by the creator.