Cloudera Data Platform provides two complementing services for data processing. Cloudera Data Engineering (CDE) for preparing and organising data to be consumed by Data Scientists in Cloudera Machine Learning (CML). This article provides a high level example of how to call CDE from CML in Python. ## Part 0: Imports # Requests will be used to make REST requests to the CDE service # JSON is used to manage payloads and responses # REST calls are made over HTTP and use BasicAuth # cde_endpoint is set to the high level URI to the CDE cluster endpoint import requests import json from requests.auth import HTTPBasicAuth cde_endpoint='cde-<cluster>.cloudera.site' The first step is establishing the credentials required to call the API. This is done by calling the Knox proxy and requesting a token. The token is then passed using the "Authorization" header. ## Part 1: Connect to Knox service. # Retrieve the JWT token and parse it for the 'access_token' part. # Need to pass workload username and password which is set as a project ENV variable. JWT = requests.get('https://service.'+cde_endpoint+'/gateway/authtkn/knoxtoken/api/v1/token', auth=HTTPBasicAuth(os.getenv('HADOOP_USER_NAME'), os.getenv('WLPASS'))) JWT_TOKEN = json.loads(JWT.content)['access_token'] auth_header={"Authorization": "Bearer %s" %JWT_TOKEN} The second step of the process is creating a resource to upload the .py file into. ## Part 2: Create a resource. # Resources provide a location to store files and dependancies. # Files are typically staged at /app/mount data={"name":"my-resource","type":"files"} resource = requests.post('https://<virtual_prefix>.'+cde_endpoint+'/dex/api/v1/resources', json=data, headers=auth_header) resource We can then upload a file from a file stream opened locally in the CML project and passing the file over HTTP/PUT into the named resource. ## Part 3: File upload to the named resource. # Once the resource is created, it can host several files. # In this example we are opening a simple python file in the project folder. files = {'file':open('/home/cdsw/helloworld.py', 'rb')} upload = requests.put('https://<virtual_prefix>.'+cde_endpoint+'/dex/api/v1/resources/my-resource/helloworld.py', files=files, headers=auth_header) upload The job then needs to be defined, this sets the type of job "spark" and links to the relevant resources. ## Part 4: Define the Job and dependancies. # Create a job definition, define its name, the resource we will use and load the file as part of the spark process. data={"name":"my-job","type":"spark","mounts":[{"resourceName":"my-resource"}],"spark":{"file":"helloworld.py"}} job = requests.post('https://<virtual_prefix>.'+cde_endpoint+'/dex/api/v1/jobs', json=data, headers=auth_header) job Finally, we can run or schedule the job. ## Part 5: Run the named Job run = requests.post('https://<virtual_prefix>.'+cde_endpoint+'/dex/api/v1/jobs/my-job/run', headers=auth_header) run.content To validate the job status, we can list the jobs running on the given cluster. ## Part 6: List all Jobs joblist = requests.get('https://<virtual_prefix>.'+cde_endpoint+'/dex/api/v1/jobs', headers=auth_header) joblist.content ... View more

Cloudera Machine Learning provides a number of methods of connecting to other CDP services and experiences such as a Cloudera Data Warehouse. In this post, we will connect using Python and the Impyla library, as well as using the embedded  Cloudera Data Visualization. Using Impyla  Within Cloudera Machine Learning, create a new project and set the language to Python 3.6. The connection details are available from the Data Warehouse console by copying the JDBC connection details which will look like. jdbc:impala://coordinator-aws-2-impala-prod.env-j2ln9x.dw.ylcu-atmi.cloudera.site:443/default;AuthMech=3;transportMode=http;httpPath=cliservice;ssl=1;UID=<workload username>;PWD=<workload password> Use the following Python code to install Impyla and configure a connection: !pip3 install impyla==0.16a3 USERNAME='<workload username>' IMPALA_HOST='coordinator-aws-2-impala-prod.env-j2ln9x.dw.ylcu-atmi.cloudera.site' IMPALA_PORT='443' from impala.dbapi import connect conn = connect(host=IMPALA_HOST, port=IMPALA_PORT, auth_mechanism='LDAP', user=USERNAME, password=os.environ['PASS'], use_http_transport=True, http_path='/cliservice', use_ssl=True) cursor = conn.cursor() cursor.execute('show databases') for row in cursor: print(row) Note: The PASS variable is an Environment variable set in the Project settings under the Advanced tab. This does not protect your password but will mitigate the risk of it being copied into a version control service. Using Visual Applications Create a  Cloudera Data Visualization App by following the instructions at Accessing Data Visualization in CML. Log out as your default user and log back into  Cloudera Data Visualization using the local admin user account. Note: You can raise a support request if you don't have access to this. Add a new connection under Basic settings using the following parameters. Connection Name: Name your Connection Hostname or IP Address:  Use the hostname from the JDBC string Port #: Use the SSL port of 443 Username: CDP Workload Username Password: CDP Workload Password Under Advanced Settings, set the following parameters. Connection Type: HTTP HTTP path: /cliservice Socket Type: SSL  Test the connection. ... View more

Cloudera Machine Learning provides support for Python3. It is very straightforward to connect a session with an operational database. Provision an Operational Database Log into a CDP instance Select  Operational Database Select   Create Database Choose the Cloud environment Provide a unique name for the database Click  Create Database Once the database has started, make a copy of the Phoenix (Thin) JDBC URL. This will be used as the connection string. Create a Machine Learning Project Within your Cloudera Machine Learning (CML) workspace, create a new project. Provide a name, and choose a blank initial setup. Create a session, and install phoenixdb using the command: !pip3 install phoenixdb Create a new Python file and paste the following code into the notebook. Import the required dependencies import phoenixdb import io import json Setup the parameters required to establish the connection with ODB. Refer to the Thin client details. opts = {} opts['authentication'] = 'BASIC' opts['serialization'] = 'PROTOBUF' opts['avatica_user'] = 'xxxxxxxx' opts['avatica_password'] = 'xxxxxxxx' database_url = 'https://<the jdbc url copied from the ODB console>/' TABLENAME = "us_population" conn = phoenixdb.connect(database_url, autocommit=True,**opts) For the URL, remove everything before the https and remove the parameters at the end, while retaining any path details. Example: https://<server>/<instance name>/cdp-proxy-api/avatica/ Create the table into which to insert the data curs = conn.cursor() query = """ CREATE TABLE IF NOT EXISTS """+TABLENAME+""" ( state CHAR(2) NOT NULL, city VARCHAR NOT NULL, population BIGINT CONSTRAINT my_pk PRIMARY KEY (state, city)) """ curs.execute(query) Bulk insert a set of data, using nested arrays for each record, and executing multiple upserts. sql = "upsert into " + TABLENAME + \ " (state ,city, population) \ values (?,?,?)" data =[['NY','New York',8143197], ['CA','Los Angeles',3844829], ['IL','Chicago',2842518], ['TX','Houston',2016582], ['PA','Philadelphia',1463281], ['AZ','Phoenix',1461575], ['TX','San Antonio',1256509], ['CA','San Diego',1255540], ['TX','Dallas',1213825], ['CA','San Jose',912332]] results = curs.executemany(sql,data) Finally, run a query to return an aggregated group-by and return as a Dictionary object. curs = conn.cursor(cursor_factory=phoenixdb.cursor.DictCursor) query = """SELECT state as "State",count(city) as "City Count",sum(population) as "Population Sum" FROM us_population GROUP BY state ORDER BY sum(population) DESC""" curs.execute(query) print(curs.fetchall()) When the above is run in a session, it will return the following results. [{'State': 'NY', 'City Count': 1, 'Population Sum': 8143197}, {'State': 'CA', 'City Count': 3, 'Population Sum': 6012701}, {'State': 'TX', 'City Count': 3, 'Population Sum': 4486916}, {'State': 'IL', 'City Count': 1, 'Population Sum': 2842518}, {'State': 'PA', 'City Count': 1, 'Population Sum': 1463281}, {'State': 'AZ', 'City Count': 1, 'Population Sum': 1461575}] This example is based on the post: Phoenix in 15 minutes or less ... View more

For a recent project I was tasked with configuring DBeaver to connect to Phoenix running in an instance of the CDP Datahub. CDP provides a means of creating a Datahub for running an Operational Database (HBase) and using JDBC via Phoenix to query it. Let's start a Datahub from an Operational Database template. Provision an Operational Database Log into a CDP Instance Select Data Hub Clusters Select Create Data Hub Choose the Cloud environment Choose the template 7.2.2 - Operational Database with SQL Provide a unique name for the cluster Click Provision Cluster   This will start a Datahub cluster running HBase and Phoenix as well as all of the security dependencies provided by SDX. For example, Knox - this will be important when connecting to our instance.   Once the cluster has started we need to collect some configuration details. This is best done on the Datahub information page and in the Cloudera Manager Console. We will use the Phoenix thin driver, and this requires a JDBC string of the form.   jdbc:phoenix:thin:url=https://<knox endpoint>:443/<cluster name>/cdp-proxy-api/avatica/;serialization=PROTOBUF;authentication=BASIC;avatica_user=<workload username>;avatica_password=<workload password>   Once the cluster has started, select Endpoints and make a note of the Phoenix Query Server URI. It should look like this:   https://<server>/opdbtest/cdp-proxy-api/avatica/   The path details here are important as they provide the proxy and the cluster name format that we need. The next piece of information we need is the Knox server endpoint. This can be found in the Cloudera Manager Console, under Knox/ Instances. This will replace the <server> part above. The final component we need is the JAR file containing the Phoenix Thin Client, and that can be sourced from the Cloudera repository here. https://repository.cloudera.com/ Search for phoenix-queryserver-client Download the latest release. Configuring DBeaver To Install DBeaver, you can download a version from https://dbeaver.io/ In this example, we are using the OSX version. Configuration fields and terms may vary my installed type. Create. new Apache Phoenix connection to provide a baseline For the host, use the machine that Knox is running on For the port, use 443 (default https port) Provide your workload username and password Edit Driver Configuration and set Class name: org.apache.phoenix.queryserver.client.Driver URL Template: jdbc:phoenix:thin:url={host}[:{port}]/opdbtest/cdp-proxy-api/avatica/;serialization=PROTOBUF;authentication=BASIC;avatica_user={user};avatica_password={password} Add the driver JAR using add File and select the JAR downloaded from the Cloudera repository Note: Don't search for class as it may automatically discover an invalid Driver class. The Class name configuration will override this. You may need to restart DBeaver, if the class is set incorrectly. Close the configuration and test the connection. The URL uses Knox so that access control can be managed centrally. Knox takes the https:// messages and proxies them through to the backend Phoenix services automatically. A note on Operational Database Experience Shortly we will be providing an Operational Database Experience. We have significantly streamlined the provisioning of the Datahubs and publish a lot more metadata to help with configuring external clients. For example, the Maven links to the correct clients are provided directly and examples of the JDBC links are presented right in the user interface.  All of these improvements have been made to help make provisioning new instances easy, and to make connecting to those instances from applications and tools very quick. Our objective is to help you integrate CDP with your applications quickly and efficiently. We welcome your feedback on areas of our platform and documentations, which can be improved to help us with this goal. Tips If you receive 404 or 401 errors, please check you are connecting to Knox, and that the full https:// url is correct. If you receive errors related to serialisation, make sure you have serialization=PROTOBUF set. Documentation references Setting up connections with a CDP Datahub Connect to PQS through Apache Knox Connecting to Apache Phoenix Query Server using the JDBC client Connect to PQS directly   Setting up connections with CDP Operational Database Experience Cloudera Operational Database JDBC support ... View more

Apache Airflow is a popular environment for scheduling and monitoring workflows. Running workflows is useful in machine learning for automation of data acquisition or the building and monitoring of machine learning models.   This tutorial will take you through the process of setting up Airflow as an Application in CML. This can be very useful in building out testing and prototype pipelines. It also provides a mechanism for calling chains of Models or Jobs.   Create a new Python project in CML using a blank template: New Project Open up a Workspace and install Airflow, this can be scripted using the install instructions here. The following shell script and python scripts can be used to automate this process. The installation needs to be performed only once: install.py: import os os.system("./install.sh")​ install.sh: #!/bin/bash -x export AIRFLOW_HOME=~/airflow # install from pypi using pip pip3 install apache-airflow # initialize the database airflow initdb​ This will install the Airflow components into your project and will persist with the lifecycle of the project. Now the software is installed the next step is starting the process as a long-lived application. This requires a python file and shell script to start up the services. install.py: import os os.system("./start.sh")​ start.sh: #!/bin/bash -x PORT=${CDSW_APP_PORT:-8090} export AIRFLOW_HOME=~/airflow # start the web server, default port is 8090 airflow webserver -p $PORT -hn 127.0.0.1 # start the scheduler airflow scheduler # visit localhost:8080 in the browser and enable the example dag in the home page​ The PORT variable is set to the environment Application Port and started on localhost. This script can be tested in a workbench, or they can be added to an Application with the following settings. Create Application Validate that the application starts up cleanly by accessing the Applications logs: Monitor Logs Opening the application will load directly into the Airflow management application. Access Airflow UX Using the Airflow HTTP Operator it is possible to call CML Jobs directly via the CML Job API.   This is hosting Airflow within a CML container.  This deployment will be isolated to the Project and will continue to run and use resources associated to the project. This provides a good prototyping environment for building out a more complete end-to-end data engineering pipelines that combine scheduling and flow with python, shell, Scala, and Spark processes.   We would be interested to hear from you, if you apply this to a project. ... View more