3
\$\begingroup\$

I am using PySpark in Azure DataBricks to try to create a SCD Type 1.

I would like to know if this is an efficient way of doing this?

Here is my SQL table:

CREATE TABLE dbo.Countries (
  CountryId bigint IDENTITY,
  ShortName nvarchar(max) NULL,
  FullName nvarchar(max) NULL,
  CapitalCities nvarchar(max) NOT NULL,
  IsActive bit NOT NULL DEFAULT (0),
  CreatedOn datetime2 NOT NULL DEFAULT (getdate()),
  UpdatedOn datetime2 NULL,
  PRIMARY KEY CLUSTERED (CountryId)
)

Here is my Python:

# Import Python Modules
import requests
import json
import pyspark.sql.functions as F
from pyspark.sql.types import ArrayType, StructType, StructField, StringType
from pyspark.sql.functions import md5, concat_ws

# Get Random Data From API
url = 'https://raw.githubusercontent.com/mledoze/countries/master/countries.json'
response = requests.get(url)
rdd_countries = sc.parallelize([response.text])
df_countries = spark.read.option("multiline","true") \
                         .json(rdd_countries)

# Remove Unwanted Columns
df_countries = df_countries.select(F.col("name.common").alias("ShortName"), F.col("name.official").alias("FullName"), F.col("capital").alias("CapitalCities"))

# Transform CapitalCities From Map To String
df_countries = df_countries.withColumn("CapitalCities", F.concat_ws(",", F.col("CapitalCities")))

# Add MD5 Hash For Row Comparison
df_countries = df_countries.withColumn("ActiveRowHash", md5(concat_ws('|',df_countries.ShortName,df_countries.FullName,df_countries.CapitalCities)))

# Rename All Source Columns
df_countries = df_countries.select([F.col(c).alias("Source_" + c) for c in df_countries.columns])

# Show Source Data
df_countries.show()

# SQL Server Config
target_server = "x.database.windows.net"
target_database = "x"
target_username = "x"
target_password = "x"
target_table = "dbo.Countries"

# Get Target Data From SQL
df_target = spark.read.format("jdbc") \
                  .option("url", f"jdbc:sqlserver://{target_server};databaseName={target_database};") \
                  .option("dbtable", target_table) \
                  .option("user", target_username) \
                  .option("password", target_password) \
                  .option("driver", "com.microsoft.sqlserver.jdbc.SQLServerDriver") \
                  .load()

# Add MD5 Hash For Row Comparison
df_target = df_target.withColumn("CurrentRowHash", md5(concat_ws('|',df_target.ShortName,df_target.FullName,df_target.CapitalCities)))

# Rename All Target Columns
df_target = df_target.select([F.col(c).alias("Target_" + c) for c in df_target.columns])

# Show Data
df_target.show()

# DataFrame For New Rows
df_new = df_countries.join(df_target, df_countries.Source_ShortName == df_target.Target_ShortName, "leftanti")
df_new.show()

# DataFrame For Deleted Rows
df_deleted = df_target.join(df_countries, df_target.Target_ShortName == df_countries.Source_ShortName, "leftanti")
df_deleted.show()

# DataFrame For Updated Rows
df_updated = df_countries.join(df_target, (df_countries.Source_ShortName == df_target.Target_ShortName) & (df_countries.Source_ActiveRowHash != df_target.Target_CurrentRowHash), "inner")
df_updated.show()

From here I will then loop though the three DataFrames and update SQL as required.

I am aware I made an assumption about the Source ShortName being the key, but that's okay for now.

Obviously I have tested the functionality of the code and it seems to work, but is it the best practice?

\$\endgroup\$
1
\$\begingroup\$

Some observations about your sql table and data types.

In most relational databases, and definitely SQL Server, choosing an appropriate data type is important.

Selecting an inappropriate data type, or being lazy about things such as string length, can result in unexpected and simply unnecessary performance degradation. Data types should be chosen such that they can cope with anticipated data, but no more.

Why? Rows in a table are arranged in structures known as pages, each page is approximately 8K in size, both on disk and in memory. A page is the minimum amount of data read by SQL Server - to read a row or rows, the page(s) containing the row are read from disk. The more compact each data type is, the more rows can fit on a single page and the less IO is requires to read a range of rows.

You have a table named Countries - presumably this stores a list of countries, of which there are currently 195 world-wide. Bigint is not the correct data type for CountryId - unless you are expecting to require at least 2,147,483,647 rows. TinyInt would be the correct data type here, its possible values range from 0-255 and uses one byte per row, instead of 8 bytes.

The same goes for usage of datetime2 - you need to provide a precision such as datetime2(1) which would use 6 bytes of storage, the default is the maximum precision that uses 8 bytes - do you really need to store the time with accuracy of one ten-millionth of a second?

Why are all your string data types nvarchar(max)? The longest country name in the world, according to Google, happens to be "The United Kingdom of Great Britain and Northern Ireland" with 56 characters. A more suitable data type would be nvarchar(60).

Likewise, the short name is presumably shorter, and the same applies to CapitalCities.

Why bother specifying an appropriate length for (n)varchar columns? They are variable by definition so only use enough space to store each value, right?

This is correct - however there is a hidden, less obvious, overhead.

When SQL Server builds an execution plan, not only does it contain all the information required to execute the query such as the operators to use to access tables, perform joins, sorts, aggregations etc it contains various meta-data and stats about the data SQL Server expects to see. One of these is an estimate of how many bytes per row are expected to be returned.

Obviously SQL Server can't know in advance what data is going to be in a particular column until the query is executed, however everything it needs to do to execute a query has to be pre-baked in advance.

All queries that are executed are allocated a minimum amount of memory in which to run, and some operations require their own additional memory allocations.

Many data types e.g. int are of fixed length and there is no ambiguity - SQL Server knows each int column of each row will be 4 bytes, so in combination with what it knows about the number of rows estimated to be processed, it can make a good estimate of how much memory to request.

For varchar data types however, SQL Server has no idea so it makes the assumption that each column will, on average, contain 50% (plus a small overhead) of the declared size. Based on this, and how many rows expected to be processed (itself determined by the cardinality and statistics of the data), SQL Server will request enough memory to process this estimated data.

This can result in queries requesting, and getting, a disproportionality large amount of memory to run, ultimately not needed, but also not available to all other concurrently executing queries.

Additionally, choosing Nvarchar as the data type means you are expecting to require unicode characters. This may be the case, in which case that's fine as you have no choice.

However be aware that each character requires two bytes of storage on disk and in memory instead of varchar's one. This can contribute to the inflated memory requirements above, and also means fewer rows of data can fit on each 8K page, resulting in more IO to read the same number of rows compared to storing as varchar, ultimately leading to slower performing queries.

I recently presented a demonstration to colleagues of how this can impact query performance.

I compared two simple tables of customer information, one with columns such as name, address and email sized appropriately e.g. using varchar(50), and one where all were nvarchar(max).

I demonstrated various contrived operations of sorting, filtering, string operations, window functions etc on each. With both tables containing identical data, the performance of the same queries on the nvarchar(max) table was up to 600% slower, and performed over 6gb of IO compared to just 4mb.

So the takeway is to always size your columns correctly for the expected data - it's just good practice!

\$\endgroup\$
0
\$\begingroup\$

A random scattering of things:

  • CapitalCities is suspicious and suggests an improperly normalized schema. One would expect a separate table with a foreign key if a country needs to represent multiple capitals. Either that or this is just misnamed and you only need to represent one capital per country, in which case - CapitalCity.
  • Apparently you're using Microsoft TSQL which violates the SQL standard fairly flagrantly. For instance, boolean is a standard SQL type but MS forces you to use bit. If you have any choice in the matter, consider migrating to an RDBMS that respects the standard.
  • Move your global code into subroutines
  • Check for Requests failure via response.raise_for_status() rather than assuming success

Also, for "fluent" syntax like this:

df_target = spark.read.format("jdbc") \
              .option("url", f"jdbc:sqlserver://{target_server};databaseName={target_database};") \
              .option("dbtable", target_table) \
              .option("user", target_username) \
              .option("password", target_password) \
              .option("driver", "com.microsoft.sqlserver.jdbc.SQLServerDriver") \
              .load()

prefer instead

df_target = (
    spark.read.format("jdbc")
    .option("url", f"jdbc:sqlserver://{target_server};databaseName={target_database};")
    .option("dbtable", target_table)
    .option("user", target_username)
    .option("password", target_password)
    .option("driver", "com.microsoft.sqlserver.jdbc.SQLServerDriver")
    .load()
)
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.