Other data destinations

​

BigQuery

​

Basic Usage

import {run} from '@subsquid/batch-processor'
import {
  Column,
  Table,
  Types,
  Database
} from '@subsquid/bigquery-store'
import {BigQuery} from '@google-cloud/bigquery'

const dataSource = /* a chain-specific data source */

const db = new Database({
  bq: new BigQuery(),
  dataset: 'subsquid-datasets.test_dataset',
  tables: {
    TransfersTable: new Table(
      'transfers',
      {
        from: Column(Types.String()),
        to: Column(Types.String()),
        value: Column(Types.BigNumeric(38))
      }
    )
  }
})

run(dataSource, db, async ctx => {
  // ...
  let from: string = ...
  let to: string = ...
  let value: bigint | number = ...
  ctx.store.TransfersTable.insert({from, to, value})
})

​

Configuration

• bq is a BigQuery instance. When created without arguments it’ll look at the GOOGLE_APPLICATION_CREDENTIALS environment variable for a path to a JSON with authentication details.
• dataset is the path to the target dataset.

• tables lists the tables that will be created and populated within the dataset. For every field of the tables object an eponymous field of the ctx.store object will be created; calling insert() or insertMany() on such a field will result in data being queued for writing to the corresponding dataset table. The actual writing will be done at the end of the batch in a single transaction, ensuring dataset integrity.

​

Deploying to SQD Cloud

sqd secrets set GAC_JSON_FILE < creds.json

import fs from 'fs'

fs.writeFileSync('creds.json', process.env.GAC_JSON_FILE || '')

deploy:
  processor:
    env:
      GAC_JSON_FILE: $
      GOOGLE_APPLICATION_CREDENTIALS: creds.json

​

BigQuery Examples

​

BigQuery Troubleshooting

1. If you are not sure if your squid is the only app that uses sessions to access your BigQuery project, find the faulty session manually and terminate it. See Get a list of your active sessions and Terminate a session by ID.
2. DANGEROUS If you are absolutely certain that the squid is the only app that uses sessions to access your BigQuery project, you can terminate all the dangling sessions at once.
   Terminating all dangling sessions

   Run the following, replacing region-us with your dataset’s region:

   FOR session in (
     SELECT
       session_id,
       MAX(creation_time) AS last_modified_time,
     FROM `region-us`.INFORMATION_SCHEMA.SESSIONS_BY_PROJECT
     WHERE
       session_id IS NOT NULL
       AND is_active
     GROUP BY session_id
     ORDER BY last_modified_time DESC
   )
   DO
     CALL BQ.ABORT_SESSION(session.session_id);
   END FOR;
   

   You can also enable abortAllProjectSessionsOnStartup and supply datasetRegion in your database config to perform this operation at startup:

   const db = new Database({
     // ...
     abortAllProjectSessionsOnStartup: true,
     datasetRegion: 'region-us'
   })
   

   This method will cause data loss if, at the moment when the squid starts, some other app happens to be writing data anywhere in the project using the sessions mechanism.

TransfersTable: new Table(
  'transfers',
  {
    from: Column(Types.String()),
    to: Column(Types.String()),
    value: Column(Types.BigNumeric(38))
  },
  5000 // <- page size for the insert operation
)

​

Filesystem Datasets

​

Overview

• CSV files - Using @subsquid/file-store-csv
• Parquet files - Using @subsquid/file-store-parquet
• JSON files - Using @subsquid/file-store-json

​

Basic Example

import {run} from '@subsquid/batch-processor'
import {Database, LocalDest} from '@subsquid/file-store'
import {Column, Table, Types} from '@subsquid/file-store-csv'

const dataSource = /* a chain-specific data source */

const dbOptions = {
  tables: {
    TransfersTable: new Table('transfers.csv', {
      from: Column(Types.String()),
      to: Column(Types.String()),
      value: Column(Types.Numeric())
    })
  },
  dest: new LocalDest('./data'),
  chunkSizeMb: 10
}

run(dataSource, new Database(dbOptions), async (ctx) => {
  for (let block of ctx.blocks) {
    // decode the transfers of the block into {from, to, value} records
    for (let {from, to, value} of decodeTransfers(block)) {
      ctx.store.TransfersTable.write({
        from,
        to,
        value: value.toString()
      })
    }
  }
})

./data/
├── 0000000000-0007688959
│   └── transfers.csv
├── 0007688960-0007861589
│   └── transfers.csv
...
├── 0016753040-0016762029
│   └── transfers.csv
└── status.txt

​

Data Partitioning

• The internal buffer (size governed by chunkSizeMb) fills up, or
• There’s a call to setForceFlush() during batch processing

​

Storage Destinations

• Local filesystem - Using LocalDest class
• S3-compatible storage - Using S3Dest class for AWS S3, Google Cloud Storage, or other S3-compatible services