# Data Processing

## Querying File

<figure><img src="https://1374779285-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2F-MFW3x2aqEO8GF2kr3VU%2Fuploads%2FWSWwej9O15VahsepjO4g%2Fimage.png?alt=media&#x26;token=d412b105-c61f-42ac-ab17-a463de2720a6" alt=""><figcaption></figcaption></figure>

```sql
SELECT * FROM read_files(
  '${dataset_bookstore}/books-csv/export_*.csv',
  format => 'csv',
  header => 'true',
  delimiter => ';'
);
```

<figure><img src="https://1374779285-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2F-MFW3x2aqEO8GF2kr3VU%2Fuploads%2FSKmazGLZL80UHTKWvVOo%2Fimage.png?alt=media&#x26;token=ce4d31f3-4109-4f38-be85-435fdad0e481" alt=""><figcaption></figcaption></figure>

* It will be an external table with same location of csv

## Streaming

<figure><img src="https://1374779285-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2F-MFW3x2aqEO8GF2kr3VU%2Fuploads%2F8BZwDcIyMeVaYpAZSGZS%2Fimage.png?alt=media&#x26;token=b2f702b9-30a2-4318-975a-ba88f6bc4abc" alt=""><figcaption></figcaption></figure>

* either a traditional approach where you reprocess the entire dataset, each time you receive a new update to your data.
* write a custom logic to only capture those files or records that have been added since the last time an update was run.

### Structured Streaming

<figure><img src="https://1374779285-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2F-MFW3x2aqEO8GF2kr3VU%2Fuploads%2FkesktuCoL3HWjrH9P1v2%2Fimage.png?alt=media&#x26;token=d606e171-cd12-483b-841a-4c8af80803d6" alt=""><figcaption></figcaption></figure>

* It allows users to interact with ever-growing data source as if it were just a static table of records.

  So new data in the input data stream is simply treated as new rows appended to a table.

  And such a table representing an infinite data source is seen as "unbounded" table.

  As we said, an **input data stream could be a directory of files, a messaging system like Kafka, or simply a Delta table**.
* spark.readStream() to query the delta table as a stream source, which allows to process all of the data present in the table as well as any new data that arrive later.
* This creates a streaming data frame on which we can apply any transformation as if it were just a

  static data frame.&#x20;

<figure><img src="https://1374779285-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2F-MFW3x2aqEO8GF2kr3VU%2Fuploads%2FT0KrWd3qDtJ59LfWBPxu%2Fimage.png?alt=media&#x26;token=1bff515f-e6e3-4d92-90c1-37f8ac3a4825" alt=""><figcaption></figcaption></figure>

* To persist the result of a streaming query. We need to write them out to durable storage using dataframe.writeStream method. With the writeStream method, we can configure our output.

### Auto Loader

<figure><img src="https://1374779285-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2F-MFW3x2aqEO8GF2kr3VU%2Fuploads%2FXxO2tsubThAlCTrA9ors%2Fimage.png?alt=media&#x26;token=d439cba1-a586-4fe0-b6bd-c8dcd6e6bc4f" alt=""><figcaption></figcaption></figure>

* Auto loader can **scale to support real time ingestion of millions of files per hour**. And since it is based on spark structured streaming, Auto Loader **uses checkpointing to track the ingestion process and to store metadata of the discovered files**.
* **Ensures that data files are processed exactly once.** And in case of failure, Auto Loader can resume from where it left off.
* Has a specific format of **StreamReader called cloudFiles**. And in order to **specify the format of the source files, we use simply cloudFiles.format** option.
* **Detect new files as they arrive and queue them for ingestion**. Once we read the files, we **write the data into a target table using the StreamWriter, where you provide the location to store the checkpointing information**.

## Multi-Hop Architecture

<figure><img src="https://1374779285-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2F-MFW3x2aqEO8GF2kr3VU%2Fuploads%2FahhuWu6Gkr5hTDJOjHFy%2Fimage.png?alt=media&#x26;token=09645abd-dd87-4ddc-9f35-5b228ae81c98" alt=""><figcaption></figcaption></figure>

* Multi hop architecture usually consists of three layers: bronze, silver and gold.
* **Bronze table** contains **raw data ingested from various sources**. Like json files, operational databases, or Kafka Stream
* **Silver table** that **provides more refined view of our data**. For example, **data can be cleaned and filtered** at this level. And we can also **join fields from various brands table** to enrich our silver records.
* Gold table that **provides business-level aggregations**, often **used for reporting and dashboarding or even for machine learning.**

### Example

#### Bronze

<figure><img src="https://1374779285-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2F-MFW3x2aqEO8GF2kr3VU%2Fuploads%2FJWKM2j5fKjWfNPXrz383%2Fimage.png?alt=media&#x26;token=63314ddc-9e86-4e88-afbe-8e97499f0711" alt=""><figcaption></figcaption></figure>

* Load the file into order-raw table and created view

<figure><img src="https://1374779285-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2F-MFW3x2aqEO8GF2kr3VU%2Fuploads%2FVMJdKE56widAC0htcr9p%2Fimage.png?alt=media&#x26;token=306fbf73-0a0e-4c41-bb1b-68aca947c4e6" alt=""><figcaption></figcaption></figure>

<figure><img src="https://1374779285-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2F-MFW3x2aqEO8GF2kr3VU%2Fuploads%2FQD1Ew3sJmklavNAhHMaI%2Fimage.png?alt=media&#x26;token=004be69a-1ca3-4161-9570-7f906fee7a1d" alt=""><figcaption></figcaption></figure>

* Create temporary view and injected into bronze table

#### Silver

<figure><img src="https://1374779285-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2F-MFW3x2aqEO8GF2kr3VU%2Fuploads%2FkihjcAx4bnHP6Y9vD4dG%2Fimage.png?alt=media&#x26;token=a45c9560-a1e6-495f-b434-90add8e318a6" alt=""><figcaption></figcaption></figure>

* Create temporary view and join other table

<figure><img src="https://1374779285-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2F-MFW3x2aqEO8GF2kr3VU%2Fuploads%2FjPCkz4ToNyd5ds3Z5ghi%2Fimage.png?alt=media&#x26;token=7bad3815-4df2-4d80-b9c8-42e2d60998b3" alt=""><figcaption></figcaption></figure>

* inject into silver table

#### Gold

<figure><img src="https://1374779285-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2F-MFW3x2aqEO8GF2kr3VU%2Fuploads%2FNxUKbd5LjJYVFMme8B6f%2Fimage.png?alt=media&#x26;token=c2bd1459-2870-47c9-b5c0-c6bb4bfdeef2" alt=""><figcaption></figcaption></figure>

* Aggregate the table to form the view

  <figure><img src="https://1374779285-files.gitbook.io/~/files/v0/b/gitbook-x-prod.appspot.com/o/spaces%2F-MFW3x2aqEO8GF2kr3VU%2Fuploads%2Fsfp1AuLWFJ22xwdoBmyE%2Fimage.png?alt=media&#x26;token=10d97d3d-5839-4e2c-a4fe-0ab9337486da" alt=""><figcaption></figcaption></figure>
* Output as a result table
