# bigquery-kafka-connect

> kafka connect connector for Google BigQuery

Latest version **3.8.1** (published 2019-02-13) · MIT license · 0 weekly downloads

## Install

```sh
npm install bigquery-kafka-connect
pnpm add bigquery-kafka-connect
yarn add bigquery-kafka-connect
bun add bigquery-kafka-connect
```

Provides the commands `nkc-bigquery-sink`, `nkc-bigquery-source`.

## Health

**Score 15/100 (F)** — status: abandoned.

Positive: no vulnerabilities.

Warnings: low downloads; no types; no esm support.

Negative: abandoned; low maintenance score.

## Facts

| | |
|---|---|
| Version | 3.8.1 |
| Published | 2019-02-13 |
| First published | 2017-06-06 |
| Weekly downloads | 0 |
| License | MIT |
| TypeScript types | none |
| Module format | CommonJS |
| Dependencies | 4 |
| Unpacked size | 239.7 KB |
| Known vulnerabilities | 0 |
| Install scripts | no |
| Author | Holger Adam |
| Maintainers | holgeradam |
| Keywords | kafka, kafka-connect, connect, connector, google, bigquery |

## Links

- npm: https://www.npmjs.com/package/bigquery-kafka-connect
- Repository: https://github.com/nodefluent/bigquery-kafka-connect
- Homepage: https://github.com/nodefluent/bigquery-kafka-connect#readme
- Issues: https://github.com/nodefluent/bigquery-kafka-connect/issues
- npm.io page: https://npm.io/package/bigquery-kafka-connect

## Dependencies (4)

- [npm](https://npm.io/package/npm.md) ^6.7.0
- [async](https://npm.io/package/async.md) ^2.6.2
- [kafka-connect](https://npm.io/package/kafka-connect.md) ^3.7.0
- [@google-cloud/bigquery](https://npm.io/package/@google-cloud/bigquery.md) ^2.0.6

## Recent versions

- 3.8.1 (latest) — 2019-02-13
- 3.8.0 — 2019-01-22
- 3.7.0 — 2018-09-25
- 3.6.1 — 2018-03-29
- 3.6.0 — 2018-03-29
- 3.5.0 — 2018-03-19
- 3.4.0 — 2018-02-23
- 3.3.0 — 2018-02-12
- 3.2.1 — 2018-02-12
- 3.2.0 — 2017-12-05
- 3.1.0 — 2017-10-13
- 3.0.0 — 2017-08-31
- 2.1.0 — 2017-07-29
- 2.0.3 — 2017-07-19
- 2.0.2 — 2017-07-13
- … 8 more at https://npm.io/package/bigquery-kafka-connect/versions

## README

# bigquery-kafka-connect

[![Greenkeeper badge](https://badges.greenkeeper.io/nodefluent/bigquery-kafka-connect.svg)](https://greenkeeper.io/)
Kafka Connect connector for Google BigQuery

[![Build Status](https://travis-ci.org/nodefluent/bigquery-kafka-connect.svg?branch=master)](https://travis-ci.org/nodefluent/bigquery-kafka-connect)

[![Coverage Status](https://coveralls.io/repos/github/nodefluent/bigquery-kafka-connect/badge.svg?branch=master)](https://coveralls.io/github/nodefluent/bigquery-kafka-connect?branch=master)

## Use API

```
npm install --save bigquery-kafka-connect
```

### bigquery -> kafka

```es6
const { runSourceConnector } = require("bigquery-kafka-connect");
runSourceConnector(config, [], onError).then(config => {
    //runs forever until: config.stop();
});
```

### kafka -> bigquery

```es6
const { runSinkConnector } = require("bigquery-kafka-connect");
runSinkConnector(config, [], onError).then(config => {
    //runs forever until: config.stop();
});
```

### kafka -> bigquery (with custom topic (no source-task topic))

```es6
const { runSinkConnector, ConverterFactory } = require("bigquery-kafka-connect");

const bigQueryTableDescription = {
    "schema": {
        "fields": [
            { name: "id", type: "INTEGER", mode: "REQUIRED" },
            { name: "name", type: "STRING", mode: "REQUIRED" },
            { name: "info", type: "STRING", mode: "NULLABLE" }
        ]
    },
    "timePartitioning": {"type": "DAY"}
};

const etlFunc = (messageValue, callback) => {

    //type is an example json format field
    if (messageValue.type === "publish") {
        return callback(null, {
            id: messageValue.payload.id,
            name: messageValue.payload.name,
            info: messageValue.payload.info
        });
    }

    if (messageValue.type === "unpublish") {
        return callback(null, null); //null value will cause deletion
    }

    callback(new Error("unknown messageValue.type"));
};

const converter = ConverterFactory.createSinkSchemaConverter(bigQueryTableDescription, etlFunc);

runSinkConnector(config, [converter], onError).then(config => {
    //runs forever until: config.stop();
});

/*
    this example would be able to store kafka message values
    that look like this (so completely unrelated to messages created by a default SourceTask)
    {
        payload: {
            id: 1,
            name: "first item",
            info: "some info"
        },
        type: "publish"
    }
*/
```

## Use CLI
note: in BETA :seedling:

```
npm install -g bigquery-kafka-connect
```

```
# run source etl: bigquery -> kafka
nkc-bigquery-source --help
```

```
# run sink etl: kafka -> bigquery
nkc-bigquery-sink --help
```

## Config(uration)
```es6
const config = {
    kafka: {
        kafkaHost: "localhost:9092",
        logger: null,
        groupId: "kc-bigquery-test",
        clientName: "kc-bigquery-test-name",
        workerPerPartition: 1,
        options: {
            sessionTimeout: 8000,
            protocol: ["roundrobin"],
            fromOffset: "earliest", //latest
            fetchMaxBytes: 1024 * 100,
            fetchMinBytes: 1,
            fetchMaxWaitMs: 10,
            heartbeatInterval: 250,
            retryMinTimeout: 250,
            requireAcks: 1,
            //ackTimeoutMs: 100,
            //partitionerType: 3
        }
    },
    topic: "sc_test_topic",
    partitions: 1,
    maxTasks: 1,
    pollInterval: 2000,
    produceKeyed: true,
    produceCompressionType: 0,
    connector: {
        batchSize: 500,
        maxPollCount: 500,
        projectId: "bq-project-id",
        dataset: "bq_dataset",
        table: "bq_table",
        idColumn: "id"
    },
    http: {
        port: 3149,
        middlewares: []
    },
    enableMetrics: true,
    batch: {
        batchSize: 100, 
        commitEveryNBatch: 1, 
        concurrency: 1,
        commitSync: true
    }
};
```

## Native Client Config(uration)
```es6
const config = {
    kafka: {
        noptions: {
            "metadata.broker.list": "localhost:9092",
            "group.id": "kc-bigquery-test",
            "enable.auto.commit": false,
            "debug": "all",
            "event_cb": true,
            "client.id": "kc-bigquery-test-name"
        },
        tconf: {
            "auto.offset.reset": "earliest",
            "request.required.acks": 1
        }
    },
    topic: "sc_test_topic",
    partitions: 1,
    maxTasks: 1,
    pollInterval: 2000,
    produceKeyed: true,
    produceCompressionType: 0,
    connector: {
        batchSize: 500,
        maxPollCount: 500,
        projectId: "bq-project-id",
        dataset: "bq_dataset",
        table: "bq_table",
        idColumn: "id"
    },
    http: {
        port: 3149,
        middlewares: []
    },
    enableMetrics: true
};
```

---
_Source: https://npm.io/package/bigquery-kafka-connect · Machine-readable twin of the npm.io package page. Health data is recomputed on every publish._