A place for tutorials on programming and other such works.
This post shows different methods of a DataFrame.
df = some_dataframe_definition
value = df.select("SOME_COLUMN_NAME").first()[0]
df = some_dataframe_definition result_json = df.toJSON()
df = some_dataframe_definition row = df.collect()[0] #You can switch out 0 for whatever row you want.
df = some_dataframe_definition num_rows = df.count()
This post is how to use pyspark to work with Delta Tables.
For more information on Delta Lake you can refer here.
First you need to install the “delta-spark” package for whatever version you require.
pip install delta-spark==3.1.0
Setup a Spark Session.
To read delta tables you can refer to PySpark: Read From ADLS to DataFrame.
To write delta tables you can refer to PySpark: Save a DataFrame To ADLS.
from delta.tables import DeltaTable vacuum_hrs = 100 path = 'abfss://<CONTAINER>@<STORAGE_ACCOUNT>.dfs.core.windows.net/<FOLDER>/' delta_table = DeltaTable.forPath(spark, path) delta_table.vacuum(vacuum_hrs)
Impoves reads by merging small files into larger ones.
from delta.tables import DeltaTable path = 'abfss://<CONTAINER>@<STORAGE_ACCOUNT>.dfs.core.windows.net/<FOLDER>/' delta_table = DeltaTable.forPath(spark, path) delta_table.optimize().executeCompaction()
from delta.tables import DeltaTable path = 'abfss://<CONTAINER>@<STORAGE_ACCOUNT>.dfs.core.windows.net/<FOLDER>/' columns = '' delta_table = DeltaTable.forPath(spark, path) delta_table.optimize().executeZOrderBy(columns)
from delta.tables import DeltaTable
import pyspark.sql.functions as F
path = 'abfss://<CONTAINER>@<STORAGE_ACCOUNT>.dfs.core.windows.net/<FOLDER>/'
delta_table = DeltaTable.forPath(spark, path)
delta_table.delete(F.col('<MY_COL>') == '<SOME_VAL>')
#You can also use sql
delta_table.delete("column == 'some_VALUE'")
You can refer here for more properties.
You would do this if you have over the max columns that the delta lake can collect statistics on. Default value is 32.
path = 'abfss://<CONTAINER>@<STORAGE_ACCOUNT>.dfs.core.windows.net/<FOLDER>/'
skip_cols = <SOME_VALUE>
spark.sql("ALTER TABLE delta.`%s` SET TBLPROPERTIES ('delta.dataSkippingNumIndexedCols' == '%s')" % (path, skip_cols))
This post is how to use the Spark Connector to Connect to SQL Server.
Install Spark Connector
spark-mssql-connector_2.12-1.2.0.jar
Install msal
pip install msal
Connect using Azure SPN
import msal
global_token_cache = msal.TokenCache()
secret = "<GET SECRET SECURELY>"
global_spn_app = msal.ConfidentialClientApplication(
<CLIENT_ID>, Authority='https://login.microsoftonline.com/<TENANT_ID>',
client_credential=secret,
token_cache=global_token_cache,
)
result = global_spn_app.acquire_token_for_client(scopes=['https://database.windows.net//.default'])
jdbc_df = spark.read \
.format("com.microsoft.sqlserver.jdbc.spark") \
.option("url", 'jdbc:sqlserver://<SERVER_NAME>:<PORT>;database=<DATABASE>;') \
.option("query", "SELECT * FROM SOMETHING") \
.option("accessToken", result['access_token']) \
.option("encrypt", "true") \
.option("hostNameInCertificate", "*.database.windows.net") \
.load()
Connect using Domain Auth
secret = "<GET SECRET SECURELY>"
jdbc_df = spark.read \
.format("com.microsoft.sqlserver.jdbc.spark") \
.option("url", 'jdbc:sqlserver://<SERVER_NAME>:<PORT>;database=<DATABASE>;') \
.option("query", "SELECT * FROM SOMETHING") \
.option("authentication", "ActiveDirectoryPassword") \
.option("user", "<USER>@<DOMAIN>") \
.option("password", "<SECRET>") \
.load()
Connect using SQL Auth
I do not recommend SQL Auth
secret = "<GET SECRET SECURELY>"
jdbc_df = spark.read \
.format("com.microsoft.sqlserver.jdbc.spark") \
.option("url", 'jdbc:sqlserver://<SERVER_NAME>:<PORT>;database=<DATABASE>;') \
.option("query", "SELECT * FROM SOMETHING") \
.option("user", "<USER>") \
.option("password", "<SECRET>") \
.load()
This post shows you how to check that a table exists in Databricks SQL.
First you will need a spark session. Refer to PySpark: Create a Spark Session for more details.
if not spark._jsparkSession.catalog().tableExists("<SCHEMA>", "<TABLE>"):
print('do something')
This post is how to set the spark configs on Databricks or Synapse Notebooks.
First you will need a spark session. Refer to PySpark: Create a Spark Session for more details.
secret = 'value' #I highly suggest you get the password from the keyvault
storage_account = ''
application_id = ''
tenant_id = ''
spark.config.set('fs.azure.account.auth.type.{}.dfs.core.windows.net'.format(storage_account), 'OAuth')
spark.config.set('fs.azure.account.oauth.provider.type.{}.dfs.core.windows.net'.format(storage_account), 'org.apache.hadoop.fs.azurebfs.oauth2.ClientCredsTokenProvider')
spark.config.set('fs.azure.account.oauth2.client.id.{}.dfs.core.windows.net'.format(storage_account), application_id)
spark.config.set('fs.azure.account.oauth2.client.secret.{}.dfs.core.windows.net'.format(storage_account), secret)
spark.config.set('fs.azure.account.oauth2.client.endpoint.{}.dfs.core.windows.net'.format(storage_account), 'https://login.microsoftonline.com/{}/oauth2/token'.format(tenant_id))
If you are running in Databricks you could add them to cluster start. Although I recommand doing it in a notebook instead.
spark.hadoop.fs.azure.account.auth.type.<STORAGE_ACCOUNT>.dfs.core.windows.net OAuth fs.azure.account.oauth.provider.type.<STORAGE_ACCOUNT>.dfs.core.windows.net org.apache.hadoop.fs.azurebfs.oauth2.ClientCredsTokenProvider fs.azure.account.oauth2.client.id.<STORAGE_ACCOUNT>.dfs.core.windows.net <CLIENT_ID> fs.azure.account.oauth2.client.secret.<STORAGE_ACCOUNT>.dfs.core.windows.net secret fs.azure.account.oauth2.client.endpoint.<STORAGE_ACCOUNT>.dfs.core.windows.net https://login.microsoftonline.com/<TENANT_ID>/oauth2/token
This how-to is how to read from ADLS to a DataFrame.
First we need a spark Session. See PySpark: Create a Spark Session for my details on that.
Read a CSV from ADLS
path = 'abfss://my_container@my_storage_account.dfs.core.windows.net/my_folder/'
format = 'csv'
#you don't need "header" if it is not CSV
dataframe = spark.read.format(format) \
.option('header', True) \
.schema(schema) \
.load(path)
Read Parquet from ADLS
path = 'abfss://my_container@my_storage_account.dfs.core.windows.net/my_folder/' format = 'parquet'
dataframe = spark.read.format(format) \
.load(path)
Read Delta from ADLS
path = 'abfss://my_container@my_storage_account.dfs.core.windows.net/my_folder/' format = 'delta'
dataframe = spark.read.format(format) \
.load(path)
This how-to is how to save a DataFrame to ADLS
First we need a spark Session. See PySpark: Create a Spark Session for my details on that.
Then we need to create a DataFrame. See PySpark: Create a DataFrame.
Then we do the following:
You should note you don’t need all the options below. I just gave an example.
path = 'abfss://my_container@my_storage_account.dfs.core.windows.net/my_folder/'
mode = 'overwrite'
format = 'delta'
partitions = []
df.write.mode(mode).format(format).option('mergeSchema', False).partitionBy(*partitions).save(path)
This post is how to create a Spark Session
Imports
from pyspark.sql import SparkSession
Create the Spark Session
spark = SparkSession.builder.appName('pyspark_app_name').getOrCreate()
You can add any configs you wish during creation. You would add this before the “.getOrCreate()”.
You can see a list here
This post is how to create a DataFrame in pyspark.
First we need a spark Session. See PySpark: Create a Spark Session for my details on that.
Next we need to import
from pyspark.sql import Row from pyspark.sql.types import StringType, DecimalType, TimestampType, FloatType, IntegerType, LongType, StructField, StructType
Then you create the schema
schema = StructType([
StructField('id', IntegerType()),
.....
])
data = [Row(id=1)]
Create the DataFrame
df = spark.createDataFrame(data, schema=schema)
If you want to use a JSON file to build your schema do the following
import json
from pyspark.sql.types import StructType
data = {
"fields": [
{
"metadata": {},
"name": "column_a",
"nullable": false,
"type": "string"
}
],
"type": "struct"
}
json_schema = json.loads(data)
table_schema = StructType.fromJson(dict(json_schema))
df = spark.createDataFrame(data, schema=table_schema)
In this tutorial I will show you how to use Kerberos/SSL with Spark integrated with Yarn. I will use self signed certs for this example. Before you begin ensure you have installed Kerberos Server and Hadoop.
This assumes your hostname is “hadoop”
cd /etc/security/keytabs/ sudo kadmin.local #You can list princepals listprincs #Create the following principals addprinc -randkey spark/hadoop@REALM.CA #Create the keytab files. #You will need these for Hadoop to be able to login xst -k spark.service.keytab spark/hadoop@REALM.CA
sudo chown root:hadoopuser /etc/security/keytabs/* sudo chmod 750 /etc/security/keytabs/*
Go to Apache Spark Download and get the link for Spark.
wget http://apache.forsale.plus/spark/spark-2.4.4/spark-2.4.4-bin-hadoop2.7.tgz tar -xvf spark-2.4.4-bin-hadoop2.7.tgz mv spark-2.4.4-bin-hadoop2.7 /usr/local/spark/
sudo nano ~/.bashrc #Ensure we have the following in the Hadoop section export HADOOP_CONF_DIR=/usr/local/hadoop/etc/hadoop #Add the following #SPARK VARIABLES START export SPARK_HOME=/usr/local/spark export PATH=$PATH:$SPARK_HOME/bin export LD_LIBRARY_PATH=$HADOOP_HOME/lib/native:$LD_LIBRARY_PATH #SPARK VARIABLES STOP source ~/.bashrc
cd /usr/local/spark/conf
mv spark-defaults.conf.template spark-defaults.conf
nano spark-defaults.conf
#Add to the end
spark.master yarn
spark.yarn.historyServer.address ${hadoopconf-yarn.resourcemanager.hostname}:18080
spark.yarn.keytab /etc/security/keytabs/spark.service.keytab
spark.yarn.principal spark/hadoop@REALM.CA
spark.yarn.access.hadoopFileSystems hdfs://NAMENODE:54310
spark.authenticate true
spark.driver.bindAddress 0.0.0.0
spark.authenticate.enableSaslEncryption true
spark.eventLog.enabled true
spark.eventLog.dir hdfs://NAMENODE:54310/user/spark/applicationHistory
spark.history.fs.logDirectory hdfs://NAMENODE:54310/user/spark/applicationHistory
spark.history.fs.update.interval 10s
spark.history.ui.port 18080
#SSL
spark.ssl.enabled true
spark.ssl.keyPassword PASSWORD
spark.ssl.keyStore /etc/security/serverKeys/keystore.jks
spark.ssl.keyStorePassword PASSWORD
spark.ssl.keyStoreType JKS
spark.ssl.trustStore /etc/security/serverKeys/truststore.jks
spark.ssl.trustStorePassword PASSWORD
spark.ssl.trustStoreType JKS
kinit -kt /etc/security/keytabs/spark.service.keytab spark/hadoop@REALM.CA klist hdfs dfs -mkdir /user/spark/ hdfs dfs -mkdir /user/spark/applicationHistory hdfs dfs -ls /user/spark
$SPARK_HOME/sbin/start-history-server.sh
$SPARK_HOME/sbin/stop-history-server.sh
I used a lot of different resources and reference material on this. Below are just a few I used.
https://spark.apache.org/docs/latest/running-on-yarn.html#configuration
https://spark.apache.org/docs/latest/security.html
This tutorial will guide you through configuring PySpark on Eclipse.
First you need to install Eclipse.
You need to add “pyspark.zip” and “py4j-0.10.7-src.zip” to “Libraries” for the Python Interpreter.
Next you need to configure the Environment variables for PySpark.
Test that it works!
from pyspark import SparkConf, SparkContext
from pyspark.sql import SparkSession
def init_spark():
spark = SparkSession.builder.appName("HelloWorld").getOrCreate()
sc = spark.sparkContext
return spark,sc
if __name__ == '__main__':
spark,sc = init_spark()
nums = sc.parallelize([1,2,3,4])
print(nums.map(lambda x: x*x).collect())
This tutorial will guide you through installing PySpark StandAlone on Windows for development.
You need to have python 3 installed. You can get it here.
Go to Apache Spark and download the latest version and package “Pre-built for Apache Hadoop 2.7 and later”. Download spark-2.4.1-bin-hadoop2.7.tgz.
You will need 7 Zip to open spark-2.4.1-bin-hadoop2.7.tgz.
You need to ensure you have Java 8 install.
Once you have installed 7 Zip you can extract spark into C:\spark\ directory. The directory structure will look like this c:\spark\spark-2.4.1-bin-hadoop2.7\
Download the winutils.exe and put to C:\spark\spark-2.4.1-bin-hadoop2.7\bin\
You the following commands to set your Spark specific ENV variables. The “-m” option means all users. You can either use that or not. If you don’t then it adds for current user only.
setx -m SPARK_HOME C:\spark\spark-2.4.1-bin-hadoop2.7 setx -m HADOOP_HOME C:\spark\spark-2.4.1-bin-hadoop2.7
You also want to add the following to the “Path” env variable “;C:\spark\spark-2.4.1-bin-hadoop2.7\bin”
Open a command prompt and type the following. The –master parameter is used for setting the master node address. local[2] is to tell Spark to run locally on 2 cores.
pyspark --master local[2]
You can then test that it is working by running the following code.
words = sc.parallelize (
["scala",
"java",
"hadoop",
"spark",
"hbase",
"spark vs hadoop",
"pyspark",
"pyspark and spark"]
)
counts = words.count()
print("Number of elements in RDD -> %i" % (counts))
I used this as a guide. https://medium.com/@GalarnykMichael/install-spark-on-windows-pyspark-4498a5d8d66c
You must be logged in to post a comment.