DEV Community: Thirumurthi S

Mounting Azure ADLS to Azure Databricks

Thirumurthi S — Mon, 14 Feb 2022 02:13:52 +0000

Pre-requisites

Azure ADLS Storage account already setup to understand this blog further
- Create a Service Principal
- Create key vault
  - Store the Tenant Id, Client Id and the Secret in the key vault
  - Key vault is a key value store, in this case provide different key name to store the values
In order to understand below block a knowledge of Databricks will help better.

How to add the key vault to the Azure Databricks Secret Scopes

Login to Azure Portal, launch the Databricks Workspace
From the Databricks workspace, in the address bar of browser append #secrets/createScope to the URL address and click enter to navigate to Secret Scope form:
In the Scope screen fill the fields as below
- Scope Name, fill in the scope name (any name example “db-app-demo-scope”.
- DNS Name fill the key vault DNS name
- Resource Id fill the Key vault Resource Id name.
Hit Create button

Note: To fill the DNS Name and Resource Id open the Azure key vault in sperate browser, copy the DNS name and Resource id.

Reference Link

Using databricks-cli to view the created secret scope.

Note: Databricks-cli is applicable only in the Cloud version or paid version and not available in community edition.

Once the secret is created, we can view the scope information and associated key vault using databricks-cli. Below cli command

  databricks secrets list-scopes --profile  my-cluster

Impacts of Service Principal renewal to Databrick ADLS mount

Before diving into the code details, lets see impacts of what happens when already mounted Databricks ADLS Mount when the Service Principal is renewed.

### Impact of renewing Service Principal on Mounted ALDS storage in Databricks

- We developed a Databricks job for our business requirement and the job uses the mounted ADLS storage to access the orc file for processing.
- To adhere to enterprise security policy after N days we renewed the Service Principal where new secret was created. 
- After renewal we get below exception and the jobs where failing.

response '{"error":"invalid_client","error_description":"AADSTS7000215: Invalid client secret is provided.

This blog is based my stack-overflow question.

Mounting the ADLS Storage in Databricks workspace

Databricks has already set with the secret scope
Below code uses the scope to access the key vault and configures the Spark session.

Note: Below code can be copy pasted into a single command let within the Databricks notebook and executed.


scopename = "db-app-demo-scope"                         # sample name
storage_acct_name = "app-storage-accnt-name"     # sample name
container_name = "mycontainer"                                 # sample name

# since the scope is set to a key vault earlier we can use the scope directly here
# The key vault created and set as a scope in databricks work space, will contain the 
# service principal created app(client)id, directory(tenant)id, and stored secret value.

# Note: we need to provide appropriate name below

app_or_client_Id=dbutils.secrets.get(scope=scopename,key="name-of-the-key-from-keyvault-referring-appid")
tenant_or_directory_Id=dbutils.secrets.get(scope=scopename,key="name-of-key-from-keyvault-referring-TenantId")
# below will be the secret created within the service principle either in portal or using az cli
secretValue=dbutils.secrets.get(scope=scopename,key="name-of-key-from-keyvaut-referring-Secretkey")

# Define Spark config dictionary for mounting to DBFS to ADLS via service principal

configs = {"fs.azure.account.auth.type": "OAuth",
          "fs.azure.account.oauth.provider.type": "org.apache.hadoop.fs.azurebfs.oauth2.ClientCredsTokenProvider",
          "fs.azure.account.oauth2.client.id": app_or_client_Id),
          "fs.azure.account.oauth2.client.secret": secretValue,
          "fs.azure.account.oauth2.client.endpoint": f"https://login.microsoftonline.com/{tenant_or_directory_Id}/oauth2/token"}

# mount to the container for ahm
mountPnt = "/mnt/my-storage/demo-app"

# Below command can be used for unmounting the container
# If the container is already mount point with that name, we simply unmount here .

dbutils.fs.unmount(mountPnt)

# only matching mountPnt will be created or checked

if not any(mount.mountPoint == mountPnt for mount in dbutils.fs.mounts()):
  print(f"Mount {mountPnt} to DBFS")
  dbutils.fs.mount(
    # pass in the container name 
    source = f"abfss://{container_name}@{storage_acct_name}.dfs.core.windows.net/",
    mount_point = mountPnt,
    extra_configs = configs)
else:
  print(f"Mount point {mountPnt} already mounted.")

# to test and list the mount that was created 
%fs ls /mnt/my-storage/demo-app

Accessing the mount point in Databricks notebook using magic command`%sql` query

Say if we have a orc file with some data content, we can use the sql query and access the mount point directly to query data.

# note I had a python note book and used the below sql magic command

%sql select * from orc.`/mnt/my-storage/demo-app/orc/demofile.orc`

Wildcard support, say if we have a folder with orc file we can use * like below

%sql select * from orc.`/mnt/my-storage/demo-app/orc/*`

Deploying Apache Artemis as container

Thirumurthi S — Mon, 31 Jan 2022 02:18:21 +0000

Steps

Building Apache Artemis Image from Github
- Clone source code from apache-artemis github project.
- Build distribution locally using maven for version 2.17.0.
- Build Docker image using locally generated artifact.
Push Image to ACR
- Tag the Docker images and push to Azure Container Registry (ACR)
Running image in Kubernetes Cluster
- Create a AKS cluster and namespace
- Use deployment manifest file to run the image
- Expose service of type load-balancer for deployment
Build client to connect to Broker
- Spring-boot client (producer/consumer) to connect to MQ Broker
Development requires below software to be installed
- Docker - 20.10+
- Kubectl - 1.20.4+ (client)

Building Docker Image

# Clone the project to specific directory
$ git clone https://github.com/apache/activemq-artemis

# fetch all tags 
$ git fetch --all --tags 

# checkout the tag to local branch
$ git checkout tags/2.17.0 –b localversion-2.17.0

# navigate and build 2.17.0 distrubiton
$ cd artemis-distribution
$ mvn  -Dskip.test=true install

// check the target directory for created file 
$ ls –lrht artemis-distribution/target/apache-artemis-2.17.0-bin/apache-artemis-2.17.0

// build the docker related files
$ ./prepare-docker.sh --from-local-dist --local-dist-path ../artemis-distribution/target/apache-artemis-2.17.0-bin/apache-artemis-2.17.0

// check docker/Dockerfile-centos file in ./activemq-artemis/artemis-distribution/target/apache-artemis-2.17.0-bin/apache-artemis-2.17.0 and use below
$ docker build -f ./docker/Dockerfile-centos -t artemis-centos . 

// Run container locally to see if the image is starting up correctly
$ docker run --rm –it –p 61616:61616 –p 8161:8161 artemis-centos

Pushing image to ACR (Azure Container Registry)

# Login to Azure 
$ az login
$ az account set -s <aks-subscription-name>
$ az acr login --name <acr-name>

# verify the image locally
$ docker pull <acr-name>.azurecr.io/<project>/artemis/v1:latest
$ docker run --rm -it -p 61616:61616 -p 8161:8161 <acr-name>.azurecr.io/<project>/artemis/v1

Create namespace, Deployment manifest file for Artemis

# Command to create  namespace
$ kubectl create namespace artemis-demo

# Add label to the namespace created
$ kubectl label namespace artemis-demo name=artemis-demo

# command used to change the default namespace is artemis-demo
$ kubectl config --current-context --namespace=artemis-demo

# navigate to the file location of artemis-deploy.yaml, issue below command
# the --namespace is not required if the first command was executed.
$ kubectl apply –f artemis-deploy.yaml --namepsace=artemis-demo
 
# the deployment and service status can be viewed with below command
$ kubectl get deployment,service -o wide

# to get the pod status and pod name
$ kubectl get pod

# the logs on the pods
$ kubectl logs pod/<pod-name-from-above-command>

Validate the service endpoint to access the Artemis console

# After deployment – use below command
$ kubectl get endpoint

# Display the service status, check if the external IP is created
$ kubect get service/artemis-server-lb

# Describe the service and fetch the LoadBalancer Ingress IP
$ kubectl describe service/artemis-server-lb | grep Load

apiVersion: apps/v1 
kind: Deployment
metadata:
  name: artemis-broker-deployment
  labels:
     app: artemis-broker
     type: artemis
     environment: dev-poc
spec:
  template:
     metadata:
        name: artemis-broker
        labels:
           app: artemis-broker
           type: artemis
     spec:
        containers:
        - name: artemis-server
          image: <azureacrname>.azurecr.io/<project>/artemis/v1   #will use the latest version.
  replicas: 1
  selector:    # This specifies which pods to be replicated 
      matchLabels:
         type: artemis
---
apiVersion: v1
kind: Service
metadata:
  name: artemis-server-lb
  labels:
    app: artemis-server-lb
  annotations:
    service.beta.kubernetes.io/azure-load-balancer-internal: "true"
spec:
  type: LoadBalancer
  ports:
    - name: tcp-acceptor
      port: 61616
      targetPort: 61616
      nodePort: 30616
      protocol: TCP
    - name: amqp-acceptor
      port: 5672
      targetPort: 5672
      nodePort: 30567
      protocol: TCP
    - name: web-console
      port: 8161
      targetPort: 8161
      nodePort: 30816
      protocol: TCP
  selector:
    app: artemis-broker
    type: artemis

Spring client code to access the containerized Artemis Queue

application.properties

brokerUrl = amqp://<external-ip-fromservice>:5672
artemisUser = artemis
artemisPassword = artemis

connectionCount=3

server.port=8085

spring.h2.console.enabled=true
spring.datasource.platform=h2
spring.datasource.url=jdbc:h2:mem:test

Rest controller, to send message

package camel.amqp.Producer;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.stereotype.Component;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

@RestController
@Component
public class ArtemisProducer {

    @Autowired
    JmsTemplate jmsTemplate;

    @PostMapping(value = "/artemis/sendMessage")
    public void sendMessage(
        @RequestParam(name = "message", required = true) final String[] message){
            jmsTemplate.convertAndSend("test.queue", message);
    }   
}

configuration class

package camel.amqp;

import org.apache.camel.component.jms.JmsConfiguration;
import org.apache.qpid.jms.JmsConnectionFactory;
import org.messaginghub.pooled.jms.JmsPoolConnectionFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.stereotype.Component;

@Component
public class AmqpJmsPoolConectionFactoryConfig {


    @Value("${artemisUser}")
    private String userName;

    @Value("${artemisPassword}")
    private String password;

    @Value("${brokerUrl}")
    private String brokerUrl;

    @Value("${connectionCount}")
    private int connectionCount;

    @Bean
    public JmsConnectionFactory jmsConnectionFactory() {
        JmsConnectionFactory jmsConnectionFactory = new JmsConnectionFactory(userName, password, brokerUrl);
        return jmsConnectionFactory;
    }

    @Bean(initMethod="start",destroyMethod="stop")
    public JmsPoolConnectionFactory jmsPoolConnectionFactory() {
        JmsPoolConnectionFactory jmsPoolConnectionFactory = new JmsPoolConnectionFactory();
        jmsPoolConnectionFactory.setMaxConnections(connectionCount);
jmsPoolConnectionFactory.setConnectionFactory(jmsConnectionFactory());
            return jmsPoolConnectionFactory;
        }

    @Bean
    public JmsConfiguration jmsConfiguration(@Value("${connectionCount}") int connectionCount){
        JmsConfiguration jmsConfiguration = new JmsConfiguration();
        jmsConfiguration.setConcurrentConsumers(connectionCount);       jmsConfiguration.setConnectionFactory(jmsPoolConnectionFactory());
        return jmsConfiguration;
    }

    @Bean
    public JmsTemplate jmsTemplate() {
        return new JmsTemplate(jmsConnectionFactory());
    }
}

Entry point to Spring Boot Application

package bootcamelamqp;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class CamelAmqpApplication {

    public static void main(String[] args) {
        SpringApplication.run(CamelAmqpApplication.class, args);
    }
}