Data export - Grain Analytics

This example builds a Node.js script that pages through the events of a date range and writes them to a CSV file and a JSON file. The script waits between pages to stay inside the rate limit of your plan, retries on 429 after Retry-After seconds, and runs from cron. The Query API needs the Builder plan or higher and an API key with the Query API permission. Read Query API.

Files

File	Holds
`src/grain-api.ts`	The API client and the retry logic.
`src/export.ts`	The script. It pages through the events and writes both files.
`.env`	`GRAIN_TENANT_ID`, `GRAIN_API_KEY`, and the pace of your plan.

The script uses the global fetch of Node.js 18 or newer and has no runtime dependency.

mkdir grain-data-export && cd grain-data-export
npm init -y
npm install -D typescript tsx @types/node

Put the tenant alias, the key, and the pace in .env.

GRAIN_TENANT_ID=your-tenant-id
GRAIN_API_KEY=$GRAIN_API_KEY
GRAIN_REQUESTS_PER_MINUTE=2

Set GRAIN_REQUESTS_PER_MINUTE to the per-minute limit of your plan. Read the pace table below.

The API client

send makes one request. On 429 it waits Retry-After seconds and retries, up to MAX_RETRIES times. On a network error it waits with exponential backoff. On any other error status it throws at once, because a retry cannot change a 401, a 403, or a 404.

// src/grain-api.ts
const BASE_URL = 'https://queryapis.grainql.com/v1/api/query';
const MAX_RETRIES = 3;

export interface QueryRequest {
  event?: string;
  after?: string;
  before?: string;
  filterSet?: Array<{ property: string; comparison: string; value: unknown }>;
  pagination?: { offset: number; size: number };
}

export interface GrainEvent {
  eventName: string;
  userId: string;
  eventTs: string;
  properties: Record<string, unknown>;
  eventDate: string;
  insertId: string;
}

export class GrainApiError extends Error {
  constructor(public status: number, message: string) {
    super(message);
  }
}

const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));

async function send<T>(path: string, init: RequestInit): Promise<T> {
  let backoffMs = 1000;
  for (let attempt = 1; ; attempt++) {
    let response: Response;
    try {
      response = await fetch(`${BASE_URL}${path}`, init);
    } catch (error) {
      if (attempt > MAX_RETRIES) throw error;
      await sleep(backoffMs);
      backoffMs *= 2;
      continue;
    }
    if (response.ok) return response.json();

    const body = await response.json().catch(() => ({ error: response.statusText }));
    if (response.status === 429 && attempt <= MAX_RETRIES) {
      const retryAfter = Number(response.headers.get('Retry-After')) || 60;
      console.error(`429 received. Waiting ${retryAfter} seconds.`);
      await sleep(retryAfter * 1000);
      continue;
    }
    throw new GrainApiError(response.status, body.error);
  }
}

export function createGrainApi(apiKey: string, tenantId: string) {
  const headers = { 'Content-Type': 'application/json', 'X-API-Key': apiKey };
  return {
    queryEvents: (request: QueryRequest) =>
      send<GrainEvent[]>(`/${tenantId}`, {
        method: 'POST',
        headers,
        body: JSON.stringify(request),
      }),
    countEvents: (request: Omit<QueryRequest, 'pagination'>) =>
      send<{ count: number }>(`/count/${tenantId}`, {
        method: 'POST',
        headers,
        body: JSON.stringify(request),
      }).then((result) => result.count),
    listEventNames: () => send<string[]>(`/events/${tenantId}`, { headers }),
  };
}

The script

The script counts the events first, then reads pages of 1,000 with a growing offset. Between two requests it waits 60 / GRAIN_REQUESTS_PER_MINUTE seconds. The CSV has one column per event field and one column per property key that appears in the export.

// src/export.ts
import { writeFileSync } from 'node:fs';
import { createGrainApi, type GrainEvent, type QueryRequest } from './grain-api';

const api = createGrainApi(process.env.GRAIN_API_KEY!, process.env.GRAIN_TENANT_ID!);
const PAGE_SIZE = 1000;
const PACE_MS = (60 / Number(process.env.GRAIN_REQUESTS_PER_MINUTE ?? 2)) * 1000;

const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));

async function exportAllEvents(request: Omit<QueryRequest, 'pagination'>): Promise<GrainEvent[]> {
  const total = await api.countEvents(request);
  console.log(`${total} events to export`);
  const events: GrainEvent[] = [];
  for (let offset = 0; offset < total; offset += PAGE_SIZE) {
    await sleep(PACE_MS);
    const page = await api.queryEvents({ ...request, pagination: { offset, size: PAGE_SIZE } });
    events.push(...page);
    console.log(`${Math.min(offset + PAGE_SIZE, total)} of ${total}`);
  }
  return events;
}

function csvCell(value: unknown): string {
  if (value === null || value === undefined) return '';
  const text = typeof value === 'object' ? JSON.stringify(value) : String(value);
  return /[",\n]/.test(text) ? `"${text.replace(/"/g, '""')}"` : text;
}

function toCsv(events: GrainEvent[]): string {
  const fields = ['eventName', 'userId', 'eventTs', 'eventDate', 'insertId'] as const;
  const propertyKeys = [...new Set(events.flatMap((event) => Object.keys(event.properties ?? {})))];
  const header = [...fields, ...propertyKeys.map((key) => `properties.${key}`)];
  const rows = events.map((event) =>
    [...fields.map((field) => event[field]), ...propertyKeys.map((key) => event.properties?.[key])]
      .map(csvCell)
      .join(','),
  );
  return [header.join(','), ...rows].join('\n') + '\n';
}

function isoDate(daysAgo: number): string {
  const date = new Date();
  date.setUTCDate(date.getUTCDate() - daysAgo);
  return date.toISOString().slice(0, 10);
}

async function main() {
  const [after = isoDate(1), before = isoDate(1), event] = process.argv.slice(2);
  const request: Omit<QueryRequest, 'pagination'> = { after, before, ...(event && { event }) };
  const events = await exportAllEvents(request);
  const stem = `export-${after}-${before}`;

  writeFileSync(`${stem}.csv`, toCsv(events));
  writeFileSync(
    `${stem}.json`,
    JSON.stringify(
      { metadata: { totalEvents: events.length, exportDate: new Date().toISOString(), request }, events },
      null,
      2,
    ),
  );
  console.log(`Wrote ${stem}.csv and ${stem}.json`);
}

main().catch((error) => {
  console.error(error.message);
  process.exit(1);
});

Run the script with a start date, an end date, and an optional event name. Without arguments it exports yesterday.

npx tsx src/export.ts 2024-01-01 2024-01-31 purchase_completed

15247 events to export
1000 of 15247
2000 of 15247
...
15247 of 15247
Wrote export-2024-01-01-2024-01-31.csv and export-2024-01-01-2024-01-31.json

The JSON file holds the events with the shape that the Query API returns:

{
  "metadata": {
    "totalEvents": 15247,
    "exportDate": "2024-02-01T02:00:07.412Z",
    "request": { "after": "2024-01-01", "before": "2024-01-31", "event": "purchase_completed" }
  },
  "events": [
    {
      "eventName": "purchase_completed",
      "userId": "user_123",
      "eventTs": "2024-01-15T10:30:00Z",
      "properties": { "price": 149.0, "currency": "USD" },
      "eventDate": "2024-01-15",
      "insertId": "unique-event-id"
    }
  ]
}

To export one segment, add a filterSet to request in main. Read Filters.

const request = {
  after,
  before,
  event: 'purchase_completed',
  filterSet: [{ property: 'properties.price', comparison: 'GREATER_THAN', value: 100 }],
};

Pace and budget

An export of N events costs 1 + ceil(N / 1000) requests: one count and one request per page. Both limits of the plan apply to the same API key:

Pace: wait at least 60 / requests per minute seconds between two requests.
Budget: (requests per day - 1) * 1000 events per day, if no other client uses the key.

Plan	Requests per minute	Requests per day	`GRAIN_REQUESTS_PER_MINUTE`	Wait between pages	Events per day
Builder	2	20	`2`	30 s	19,000
Growth	10	200	`10`	6 s	199,000
Scale	100	2,000	`100`	0.6 s	1,999,000

If a date range holds more events than the daily budget, export it one day at a time on consecutive days, or narrow it with event or filterSet.

Scheduled run

Run the script from cron at a fixed time. The line below exports yesterday every day at 02:00 and appends the log to export.log.

crontab -e

0 2 * * * cd /srv/grain-data-export && npx tsx src/export.ts >> export.log 2>&1

Rules for the schedule:

Load .env in the cron environment. Cron does not read your shell profile. Use node --env-file=.env (Node.js 20.6 or newer) or export the variables in the crontab.
Leave a gap between two scheduled exports that share a key. Both count against one budget.
Make sure that a run finishes before the next one starts. A Builder export of 19,000 events takes about 10 minutes at 30 seconds per page.

Errors

Status	Body	What to do
`401`	`{"error":"Missing X-API-Key header"}` or `{"error":"Invalid API key or insufficient permissions"}`	Make sure that `GRAIN_API_KEY` is set and that the key has the Query API permission.
`403`	`{"error":"Query API requires Builder plan or higher. ..."}`	Upgrade the plan. The script stops.
`404`	`{"error":"Tenant not found"}`	Use the tenant alias from the dashboard in `GRAIN_TENANT_ID`, not the UUID.
`429`	`{"error":"Rate limit exceeded. Upgrade for higher limits."}` with `Retry-After`	The client waits `Retry-After` seconds and retries up to 3 times. If it still fails, lower `GRAIN_REQUESTS_PER_MINUTE` or run the export later.
`400`	`{"error":"Invalid date format. Use YYYY-MM-DD format."}`	Pass dates as `YYYY-MM-DD`.

Limits

Limit	Value	Where
Events per page	1,000	`pagination.size`. `PAGE_SIZE` in `src/export.ts`.
Retries per request	3	`MAX_RETRIES` in `src/grain-api.ts`.
Requests per minute	2, 10, or 100	Per API key, by plan. Read the pace table.
Requests per day	20, 200, or 2,000	Per API key, by plan. Custom plans with higher limits exist.
Memory	All events of one run	The script holds the events in memory before it writes the files. Export one day per run.

Read Custom dashboard for a Next.js page that polls the same endpoints.

​Files

​The API client

​The script

​Pace and budget

​Scheduled run

​Errors

​Limits