> For the complete documentation index, see [llms.txt](https://docs.fractalworks.io/joule/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.fractalworks.io/joule/components/pipelines/data-priming/types-of-import.md).

# Types of import

## Overview

The page describes how data imports from Parquet and CSV files can be configured, how to define tables and indexes and how these imports can be utilised within a processing pipeline.

The provided example demonstrates the integration of imported data into a stream-based analytics system, with calculations and data enrichment performed on the incoming data.

## Parquet Import

Parquet formatted files can be imported into the system.

{% hint style="info" %}
`index` cannot be created over a view
{% endhint %}

### Example

This example imports Parquet files into different schemas and tables.

Some tables are views: `us_holidays`, others are regular tables: `fxrates`, `bid_moving_averages`.

Indexes are created on specific fields but are not unique and existing tables are dropped before importing new data.

```yaml
stream:
  ...
  data import:
    parquet:
      - schema: exchange_rates
        table: fxrates
        asView: false
        files: [ 'fxrates.parquet' ]
        drop table: true
        index:
          fields: [ 'ccy' ]
          unique: false
          
      - schema: reference_data
        table: us_holidays    
        asView: true
        files: [ 'holidays.parquet' ]
        drop table: true
        
      - schema: metrics
        table: bid_moving_averages
        files: ['data/parquet/mvavgs-prime.parquet']
        drop table: true
        index:
          fields: [ 'symbol' ]
          unique: false
```

### Attributes schema

<table><thead><tr><th width="145">Attribute</th><th width="377">Description</th><th width="136">Data Type</th><th data-type="checkbox">Required</th></tr></thead><tbody><tr><td>asView</td><td>Create a view over the <code>files</code>. This will mean disk IO for every table query. Consider <code>false</code> if the table is small and accessed frequently</td><td><p>Boolean</p><p><em>Default false</em></p></td><td>false</td></tr><tr><td>files</td><td>List of <code>files</code> of the same type to be imported</td><td>String list</td><td>true</td></tr></tbody></table>

## CSV import

Data can be imported from CSV files using a supported set of delimiters. The key difference between parquet and CSV, it is possible to **control the table definition**.

Joule by default will try to create a target table based upon a sample set of data **assuming a header exists** on the first row.

### Example

This example imports a CSV file: `fxrates.csv` into the `nasdaq` table in the `reference_data` schema. It specifies the table structure, including column types.

Key settings include a custom delimiter (`;`), date and timestamp formats, sample size and no automatic detection of data types.

The table is dropped before importing and an index is created on the `symbol` field.

```yaml
stream:
  ...
  data import:
    csv:
      - schema: reference_data
        table: nasdaq
        table definition:
          nasdaq (
          ccy VARCHAR,
          name VARCHAR,
          last_sale REAL,
          net_change REAL,
          change REAL,
          market_cap DOUBLE,
          country VARCHAR,
          ipo_year SMALLINT,
          volume INTEGER,
          sector VARCHAR,
          industry VARCHAR);
        file: 'data/csv/fxrates.csv'
        delimiter: ";"
        date format: "%Y/%m%d"
        timestamp format: "%Y/%m%d"
        sample size: 1024
        skip: 0
        header: true
        auto detect: false
        drop table: true
        index:
          fields: [ 'symbol' ]
          unique: false
```

### Attributes schema

<table><thead><tr><th width="145">Attribute</th><th width="377">Description</th><th width="136">Data Type</th><th data-type="checkbox">Required</th></tr></thead><tbody><tr><td>table definition</td><td>Custom SQL table definition used when provided. This will override the <code>auto detect</code> flag</td><td><em>String</em></td><td>false</td></tr><tr><td>file</td><td>Name and path of the file to import.</td><td>String</td><td>true</td></tr><tr><td>delimiter</td><td>Field delimiter to use. Supported delimiters include <code>, | ; and space and tab</code></td><td><p>String</p><p>Default |</p></td><td>false</td></tr><tr><td>date format</td><td>User specified date format. Many formats are possible, i.e: %d/%Y/%m | %d-%m-%Y</td><td><p>String</p><p>Default: System</p></td><td>false</td></tr><tr><td>timestamp formate</td><td>User specified timestamp format. Many formats are possible, i.e: %d/%Y/%m | %d-%m-%Y</td><td><p>String</p><p>Default: System</p></td><td>false</td></tr><tr><td>sample size</td><td>Number of rows to use to determine types and table structure</td><td><p>Integer</p><p>Default: 1024</p></td><td>false</td></tr><tr><td>skip</td><td>Number of rows to skip when auto generating a table using a sample size</td><td><p>Integer</p><p>Default: 1</p></td><td>false</td></tr><tr><td>header</td><td>Flag to indicate the first line in the file contains a header</td><td><p>Boolean</p><p>Default: true</p></td><td>false</td></tr><tr><td>auto detect</td><td>Auto detect table format by taking a sample size of data. If set to <code>false</code> a table definition must be provided</td><td><p>Boolean</p><p>Default: true</p></td><td>false</td></tr></tbody></table>

## Application example

The following example primes the process with [contextual](/joule/components/contextual-data.md) and [metrics](/joule/components/processors/enrichment/metrics.md) data which is used for event [enrichment](/joule/components/processors/enrichment.md) processing.

The code defines a stream named `standardQuoteAnalyticsStream` that imports data from CSV: `nasdaq.csv` and Parquet: `mvavgs-prime.parquet` files.

It computes the `BidMovingAverage` metric, enriches the data with company info and quote metrics and outputs the data grouped by `symbol`.

The stream runs every minute, with compaction every 8 hours and flushes data every 5 minutes.

The data is processed based on event timestamps and includes table imports, metric calculations and enrichment steps.

```yaml
stream:
  name: standardQuoteAnalyticsStream
  enabled: true
  eventTimeType: EVENT_TIME
  sources: [ nasdaq_quotes_stream ]

  initialisation:
    data import:
      csv:
        - schema: reference_data
          table: nasdaq_companies
          file: 'data/csv/nasdaq.csv'
          drop table: true
          index:
            fields: [ 'Symbol' ]
            unique: true

      parquet:
        - schema: metrics
          table: bid_moving_averages
          files: ['data/parquet/mvavgs-prime.parquet']
          drop table: true
          index:
            fields: [ 'symbol' ]
            unique: false

  processing unit:
    metrics engine:
      runtime policy:
        frequency: 1
        startup delay: 1
        time unit: MINUTES

      foreach metric compute:
        metrics:
          -
            name: BidMovingAverage
            metric key: symbol
            table definition: bid_moving_averages (symbol VARCHAR, avg_bid_min FLOAT, avg_bid_avg FLOAT,avg_bid_max FLOAT,createdTimestamp TIMESTAMP)
            query:
              SELECT symbol,
              MIN(bid) AS 'avg_bid_min',
              AVG(bid) AS 'avg_bid_avg',
              MAX(bid) AS 'avg_bid_max'
              FROM quotes.nasdaq
              WHERE
              ingestTime >= epoch_ms(date_trunc('minutes',now() - INTERVAL 3 MINUTES)) AND ingestTime <= epoch_ms(now())
              GROUP BY symbol
              ORDER BY 1;
            truncate on start: false
            compaction policy:
              frequency: 8
              time unit: HOURS

    pipeline:
      - tap:
          target schema: quotes
          flush frequency: 5
          index:
            unique: false
            fields:
              - symbol
      - enricher:
          fields:
            company_info:
              by query: "select * from reference_data.nasdaq_companies where Symbol = ?"
              query fields: [ symbol ]
              with values: [ Name,Country ]
              using: JouleDB
            quote_metrics:
              by metric family: BidMovingAverage
              by key: symbol
              with values: [avg_bid_min, avg_bid_avg, avg_bid_max]
              using: MetricsDB
  emit:
    select: "symbol, Name, Country, avg_bid_min, avg_bid_avg, avg_bid_max"

  group by:
    - symbol
```