Here, will talk about querying files in Databricks.
We will learn how to extract data directly from files using Spark SQL. And different methods to review raw files contents. We will see how to use the Spark SQL to configure options for extracting data from external sources and how to use CTAS statements to create Delta Lake tables.
To query or file content, we can simply use a select statement. SELECT * FROM a file format, and we specify the file path.
SELECT * FROM file_format.`/path/to/file`
And make special note of the use of backticks (``) and not single quotes around the path. This work well with self-describing formats that have well-defined schema like JSON and parquet. However, it is not very useful with non describing formats like CSV. A path file could be a single file or we can use wildcard character to read multiple files simultaneously or simply reading the whole directory.
Of course, assuming that all of the files in the directory have the same format and schema. Here is an example to query a JSON file. As we see, it simply SELECT * FROM json, and we specify the path file around back ticks.
SELECT * FROM json.`/path/file_name.json`
When working with text-based files which include JSON, CSV, TSV and TXT format, we can use the "text" format to extract data as raw strings. This can be useful when input data could be corrupted. In this case, we extract the data as raw string and we apply custom text parsing functions to extract values from text files.
And in some cases, we need the binary representation of files content, for example, when dealing with images and unstructured data. Here we can use simply "binaryFile" as a format. And usually after extracting data from external data sources, we need to load them into the lakehouse which ensures that all of the benefits of Databricks platform can be fully leveraged.
To load data from files into Delta tables. We use CTAS statements, which is Create Table As Select query. Here we are querying data from files directly. CTAS statements automatically infer schema information from query results and do not support manual schema declaration. This means the CTAS statements are useful for external data ingestion from sources with well-defined schema such as parquet files and tables. CTAS statements also do not support specifying additional file options. And this is why this statement presents significant limitation when trying to ingest data from CSV files. For such a format that requires additional options. We need another solution that supports options.
This solution is the regular Create Table statement, but with the USING keyword. By adding the USING keyword, we specify the external data source type, for example CSV format and with any additional options. And of course, we need to specify a location to where these files are stored. That means with this command, we are always creating an external table. The table here is just a reference to the files. Unlike with CTAS statements, here there is no data moving during table creation. We are just pointing to files stored in an external location.
<aside> 💡
Moreover, these files are kept in its original format, which means we are creating here a non Delta table.
</aside>
Here is an example of creating a table using CSV external source. So again, it's not a Delta table. We are pointing to CSV files exist in an external location. And we are specifying the options for reading the files. Like the fact that there is a header presents in the files and the delimiter is a semicolon. And finally, we are providing the location to these CSV files.
Another example is to create a table using JDBC connection to refer to data in an external SQLdatabase. And we provide the necessary options like the connection string, the username and the password for this database and of course, the database table containing the data. And again, a table with external data source has a limitation.
It is not a Delta table. It means the performance and the features of Delta Lake are no more guaranteed, like time travel feature, and the guarantee that we are always reading the most recent version of the data. In addition, if we are referring to a huge database table, this also can cause performance issues.
But fortunately we have a solution for this limitation. The solution is simply to create a temporary view, referring to the external data source, and then query this temporary view to create a table using CTAS statements. In this way we are extracting the data from the external data source and load it in a Delta table. And as we can see, with CTAS statement, we can not only query files, but we can query any objectlike a temporary view in this case.
Here, we will learn how to extract data directly from files using Spark SQL on Databricks. For this demonstration. We will work on a bookstore dataset. Here we can see the schema for this bookstore dataset.
We have three tables, customers, books and orders. Let us start by running this helping notebook that will download and copy the dataset to our databricks file system.
-- Run the Copy-Datasets file from the repo provided
%run /Repos/[email protected]/Databricks-Certified-Data-Engineer-Associate/Includes/Copy-Datasets
The customer data is coming in a JSON format. Let us list the files in the customers directory.
%python
files = dbutils.fs.ls(f"{dataset_bookstore}/customers-json")
display(files)
We can see that there are six JSON files in this directory. Let us try to read them. To query a single json file, we use SELECT * FROM json, and we specify the full path for this JSON file.
SELECT * FROM json.`${dataset.bookstore}/customers-json/export_001.json`
And notice here that we are using backticks and not single quotes around the path. So we managed to read the table, and here we can see the different columns of our table, the customer's ID, the email, the profile information, which itself is JSON string.
And the last updated timestamp. And we can see also that our preview display shows all the 300 records of our source file. We can also use wildcard character to query multiple files simultaneously.
SELECT * FROM json.`${dataset.bookstore}/customers-json/export_*.json`
For example, here we are querying all the JSON files, starting with the name export_.By default, the preview display shows only the first 1000 records. In addition, we can query a complete directory of files, assuming all the files in the directory have the same format and schema.
SELECT * FROM json.`${dataset.bookstore}/customers-json`
Here we will specify simply the directory path rather than an individual file. So we have successfully read all data in this directory.
Let us see how many customers we have.
SELECT count(*) FROM json.`${dataset.bookstore}/customers-json`
So, we have 1700 customers in this dataset. When reading multiple files, it is useful to add the input_file_name function, which is a built-in Spark SQL command that records the source data file for each record. This can be especially helpful if troubleshooting problems in the source data become necessary.
SELECT
*,
input_file_name() as source_file
FROM json.`${dataset.bookstore}/customers-json`;
So here, in addition to our columns, we have also the source file column, which is really helpful.
Another interesting option here is to use the text format, which allows we to query any text based files like JSON, CSV, TSV, or TXT format.
As we can see, this load each line of the file as a row with one string column named "Value" and this can be useful when data could be corrupted. And we need to use, in such cases, some custom text parsing function to extract data. In addition, we can use binaryFile to extract the raw bytes and some metadata of files.
So, as we can see here, this gives us the path of the file, the modification time, the length and the content, which is the binary representation of the file.
Let us now switch to reading books data which is coming in CSV format. In the same way we will use the SELECT statement, but this time with the CSV format.
We managed to read the data. However, it is not well parsed!The header row is being extracted as a table row and all columns are being loaded in a single column. And it seems that this is because of the delimiter of the file, which is in our case, a semicolon instead of comma. In fact, querying files in this way works well only with self-describing formats. The formats that have well-defined schema like JSON and parquet. However, for other formats like CSV where there is no schema defined this does not work and we need another way that allows us to provide additional configuration and schema declaration. One solution is to create a table with the USING keyword. This allows us to create a table against an external sources like CSV format. So, here we need to specify the table schema. I mean, the column names and types, the file format, which is in our case, CSV.
CREATE TABLE books_csv
(book_id STRING, title STRING, author STRING, category STRING, price DOUBLE)
USING CSV
OPTIONS (
header="true",
delimeter=";"
)
LOCATION "${dataset.bookstore}/books-csv"
And whether if there is a header in the source files, and the delimiter used to separate fields, in our case, it's a semicolon. And finally, we need to specify the location to the files directory. Let's now query this table.
So we managed to read the books that are in the CSV files. And remember when working with CSV files as data source, it is important to ensure that column orders does not change if additional data files will be added to the source directory.
<aside> 💡
Spark will always load data and apply columns names and data types in the order specified during table creation.
</aside>
Let us now run Describe Extended to see some information on our table.
Here we can see that we have created an external table. And this table is not a Delta table. It's a table that's referring directly to the CSV files. It means that no data has moved during table creation. We are just pointing to files stored in an external location. In addition, all the metadata and options passed during table creation will be persisted to the metadata, ensuring that data in the location will always be read with these options.
Let us now see the impact of not having a Delta table. In fact, all the guarantees and features that we have usually when work with Delta tables we will no longer having them with external data sources like CSV. For example, Delta Lake tables guarantee that we always query the most recent version of our source data, while tables registered against other data sources like CSV may represent older cached versions. Let us add some new CSV file to our directory and see what will happen. First, let us check how many CSV files we have in the directory.
So currently we have four CSV files. Here we will use a Spark DataFrame API that allows us to write data in a specific format like CSV.
%python
(
spark.read
.table("books_csv")
.write
.mode("append")
.format("csv")
.option('header', 'true')
.option('delimeter', ';')
.save(f'{dataset_bookstore}/books-csv')
)
For this demonstration, the idea is simple. We are going to read our books table we have just created. And we are going to rewrite the table data in a new additional CSV files in the same directory. Let us now see how many CSV files in the directory.
Yes, indeed, there are extra CSV files that have been written to the directory by Spark. Now, if we calculate the number of books in our table, we should see 24 rows instead of 12.
But even with the new data has been successfully written to the table directory, we are still unable to see this new data. And this is because Spark automatically cached the underlying data in local storage to ensure that on subsequent queries, Spark will provide the optimal performance by just querying this local cache. This external CSV file is not configured to tell Spark that it should refresh this data. However, we can manually refresh the cache of our data by running the REFRESH table command.
But remember, refreshing a table will invalidate its cache, meaning that we will need to re-scan our original data source and pull all data back into memory. For a very large dataset, this may take a significant amount of time. Let us now check again the number of books in our table. And indeed, after refreshing the table, now we see that we have 24 books. So as we can see, non Delta tables have some limitations.
To create Delta tables where we load data from external sources, we use Create Table AS Select statements or CTAS statements.
CREATE TABLE customers AS
SELECT * FROM json.`${dataset.bookstore}/customres-json`;
DESCRIBE EXTENDED customers;
Here, we create and populate customer data table using data retrieved from this input query. These will extract the data from the JSON files and load them into the table customers.
From the table metadata, we can see that we are indeed creating a delta table, and it is also a managed table. In addition, we can see that schema has been inferred automatically from the query results. This is because CTAS statements automatically infer schema information from a query results and do not support manual schema declaration. This means that CTAS statements are useful for external data ingestion from sources with well-defined schema such as parquet files and tables.
CREATE TABLE books_unparsed AS
SELECT * FROM csv.`${dataset.bookstore}/books-csv`;
SELECT * FROM books_unparsed;
In addition, CTAS statements do not support specifying additional file options which presents significant limitation when trying to ingest data from CSV files. So we have successfully created a Delta table here. However, the data is not well parsed. To correct this, we need first to use a reference to the files that allow us to specify options. And this is what we are doing here by creating this temporary view that allows us to specify file options.
CREATE TEMP VIEW books_tmp_vw
(book_id STRING, title STRING, author STRING, category STRING, price DOUBLE)
USING CSV
OPTIONS (
path = "${dataset.bookstore}/books-csv/export_*.csv",
header = "true",
delimiter ";"
);
CREATE TABLE books AS
SELECT * FROM books_tmp_vw;
SELECT * FROM books
Then we will use this temporary view as the source for our CTAS statement to successfully register the Delta table. The table has been successfully created. And notice here we are retrieving only 12 records because we use the wildcard character in the path location.
Let us finally check the metadata of our Delta table.
And yes, indeed, it's a Delta table where we have extracted all the data from the CSV files and loaded them into this location.
Here, we are going to explore SQL syntax to insert and update records in Delta tables. And remember Delta technology provides ACID compliant updates to Delta tables. For this demonstration, we will continue working with our bookstore dataset. Let us first run our helping notebook to copy the dataset.
%run /Repos/[email protected]/Databricks-Certified-Data-Engineer-Associate/Includes/Copy-Datasets
And then we will use a CTAS statement to create orders delta table As Select statement from parquet files.
CREATE TABLE orders AS
SELECT * FROM parquet.`${dataset.bookstore}/orders`
Let's query this table.
As we can see, parquet files have a well-defined schema, so we managed to extract the data correctly. When writing to tables, we could be interested by completely overwriting the data in the table. In fact, there are multiple benefits to overwriting tables instead of deleting and recreating tables. For example, the old version of the table still exists and can easily retrieve all data using Time Travel. In addition, overwriting a table is much faster because it does not need to list the directory recursively or delete any files. In addition, it's an atomic operation. Concurrent queries can still read the table while we are overwriting it. And of course due to the ACID transaction guarantees, if overwriting the table fails, the table will be in its previous state.
CREATE OR REPLACE TABLE orders AS
SELECT * FROM parquet.`${dataset.bookstore}/orders`