Databricks CLI fundamentals

 

1. pip3 install databricks-cli
2. Check if installed: which databricks
3. Check version: databricks —version
4. databricks configure —token
5. databricks clusters list
6. To edit default configure: vi .databrickscfg
7. Create scope: databricks secrets create-scope  —scope demo
8. Put APP_key into the scope: databricks put —scope demo —key APP_key —string-value some-value
9. To configure password file: 
• vim password.txt (add at the end to remove newline character: set noendofline binary) use :wq to quit editing
• databricks put —scope demo —key password —binary-file password.txt
10. To delete scopes: databricks secrets delete-scope —scope demo
11. To push project to Databricks workspace and load the .whl file to dbfs:

1. To install the .whl file from CLI:
• Get cluster-id: databricks clusters get —cluster-name demo
• To install lib: databricks libraries install —cluster-id your-cluser-id —whl dbfs:/tmp/whl-name.whl
2. To export changes made in Databricks and sync with local and use :git diff weather-wheel.py to see the differences:

1. To import local changes to sync with Databricks(completely overwrite): databricks workspace import -o -l PYTHON weather-notebook.py /cli-demo/weather-notebook
2. Some other interactions with Databricks CLI:
• Start a cluster: databricks clusters start —cluster-id your-cluster-id
• List jobs: databricks jobs list
• Get job detail: databricks jobs get  —job-id job-id-number
• Run a job: databricks jobs run-now —job-id job-id-number
• Get running job detail: databricks runs get-output —run-id id-from-last-step
• To terminate(not delete) a cluster: databricks clusters delete —cluster-id your-cluster-id

To create secrets using Databricks CLI:

- databricks secrets create-scope --scope  your-scope-name

- databricks secrets put --scope your-scope-name  --key username --string-value blabla

- databricks secrets list --scope  your-scope-name

To check secrets in Databricks:

- dbutils.secrets.listScopes()

- dbutils.secrets.list('demo')

- dbutils.secrets.get(scope="demo", key="app_key")

- trick to see the key:

app_key = []

for x in dbutils.secrets.get(scope="demo", key="app_key"):

  if x is not None:

    app_key.append(x)

print("app_key:", ' '.join(app_key))

To create .whl file:

- python -m build

Spark - register table in the Hive metastore

 %sql

2

DROP TABLE IF EXISTS health_tracker_processed;

3

​

4

CREATE TABLE health_tracker_processed                        

5

USING PARQUET                

6

LOCATION "/dbacademy/$username/DLRS/healthtracker/processed"

Per best practice, we have created a partitioned table. However, if you create a partitioned table from existing data, Spark SQL does not automatically discover the partitions and register them in the Metastore.

MSCK REPAIR TABLE recovers all the partitions in the directory of a table and updates the Hive metastore. When creating a table using PARTITIONED BY clause, partitions are generated and registered in the Hive metastore. However, if the partitioned table is created from existing data, partitions are not registered automatically in the Hive metastore. User needs to run MSCK REPAIR TABLE to register the partitions. MSCK REPAIR TABLE on a non-existent table or a table without partitions throws an exception. Another way to recover partitions is to use ALTER TABLE RECOVER PARTITIONS.

health_tracker_processed = spark.read.table("health_tracker_processed")

health_tracker_processed.count()

Above returns 0 count. You have to use below to register partition.

%sql

MSCK REPAIR TABLE health_tracker_processed

then if you do count(), it will return result.

GitLab - Fork using CLI

 

Fork using CLI:

1. Clone our own project repo to local:  git clone https://gitlab.com/XXX.git
2. Add upstream repo: git remote add upstream https://gitlab.com/XXXX
3. To verify the remote has been added: git remote -v
4. Pull changes from upstream to local: git pull upstream master
5. Push to origin (Fork: our own repo): git push origin master

Reference : https://www.sitepoint.com/quick-tip-synch-a-github-fork-via-the-command-line/

But it is only click the 'Fork' button that you can create merge request to the upstream repo. So better use CLI only to sync with the upstream repo.

SQL - DB summary on count of SP, views, functions, tables

 You can use T-SQL to Count Number Of Stored Procedures, Views, Tables or Functions in a Database by using the Database INFORMATION_SCHEMA view:

SELECT COUNT(*) AS TABLE_COUNT FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_TYPE='BASE TABLE'

 

/* Count Number Of Views In A Database */

SELECT COUNT(*) AS VIEW_COUNT FROM INFORMATION_SCHEMA.VIEWS

 

/* Count Number Of Stored Procedures In A Database */

SELECT COUNT(*) AS PROCEDURE_COUNT FROM INFORMATION_SCHEMA.ROUTINES WHERE ROUTINE_TYPE = 'PROCEDURE'

 

 /* Count Number Of Functions In A Database */

SELECT COUNT(*) AS FUNCTION_COUNT FROM INFORMATION_SCHEMA.ROUTINES WHERE ROUTINE_TYPE = 'FUNCTION'

PySpark - Map values

 Sometimes you want to define the hierarchy of the data source or just want to define the importance of something data. You can use create_map() function

  src_type = {"first_important": 1, "second_important": 2, "third_important": 3}

  src_type = F.create_map([F.lit(x) for x in chain(*src_type.items())])

Then you can get result like this:

+-------------------------------------------------------------------------+
|Map(first_important-> 1, second_important-> 2, third_important-> 3)      |
+----+--------------------------------------------------------------------+

When you use it, you can do something like

df=df.withColumn("is_type",F.when((src_type[F.col("tgt_type")] \

      >src_type[F.col("src_type")]), 'True').otherwise("False"))

PySpark - Read Shape file from S3 and Mount S3 as File System

Advantages of Mounting Amazon S3 as a File System

Mounting an Amazon S3 bucket as a file system means that you can use all your existing tools 
and applications to interact with the Amazon S3 bucket to perform read/write operations on 
files and folders. Can EC2 mount Amazon S3? Using this method enables multiple Amazon 
EC2 instances to concurrently mount and access data in Amazon S3, just like a shared file 
system.

Why use an Amazon S3 file system? Any application interacting with the mounted drive 
doesn’t have to worry about transfer protocols, security mechanisms, or Amazon 
S3-specific API calls. In some cases, mounting Amazon S3 as drive on an application 
server can make creating a distributed file store extremely easy.

For example, when creating a photo upload application, you can have it store data on a fixed 
path in a file system and when deploying you can mount an Amazon S3 bucket on that fixed 
path. This way, the application will write all files in the bucket without you having to worry about
 Amazon S3 integration at the application level. Another major advantage is to enable legacy 
applications to scale in the cloud since there are no source code changes required to use an 
Amazon S3 bucket as storage backend: the application can be configured to use a local path 
where the Amazon S3 bucket is mounted. This technique is also very helpful when you want 
to collect logs from various servers in a central location for archiving.

After mounting S3 as local file system, you can use Pandas or others to access file using 
path like: 

Location = geopandas.read_file("/dbfs/mnt/bucket-name/geofactor/data/shapefilename.shp")

But in my case, since we have limitations on mounting S3 as well as permission issues 
(only Spark can ready from S3 bucket), but the file type is shape file which includes .dbf, .prj, 
.shp and .shx type files and we have to read them as a whole. so I zipped the file. So basically, 
we cannot use Spark to read this zip file. 

We have to use boto3 to read the zip file. And the work around it instead of mounting S3 is to 
read the zip file using BytesIO().
buffer = BytesIO(zip_obj.get()["Body"].read())
zipfile = ZipFile(io.BytesIO(buffer.read()))

PySpark - Two ways to use UDF

 Method 1:

put @F.udf('string')  on top of the function

Method 2:

Change python function to an UDF function:

from pyspark.sql.functions import *

from pyspark.sql.types import *

function_udf = udf(function, OutputType())

then use the UDF function in your Spark dataframe.