## 1. Project Overview & Quickstart (jitsucom/bulker) ## File: README.md > [!CAUTION] > ## ๐Ÿšจ This repository has been archived > > **Bulker has moved.** The code now lives inside the main Jitsu monorepo: > **๐Ÿ‘‰ [jitsucom/jitsu โ€” /bulker](https://github.com/jitsucom/jitsu/tree/newjitsu/bulker)** > > This repo is read-only. Please open issues and PRs in the main repo. --- # ๐Ÿšš Bulker Bulker is a tool for streaming and batching large amount of semi-structured data into data warehouses. It uses Kafka internally ## How it works? Send and JSON object to Bulker HTTP endpoint, and it will make sure it will be saved to data warehouse: * **JSON flattening**. Your object will be flattened - `{a: {b: 1}}` becomes `{a_b: 1}` * **Schema managenent** for **semi-structured** data. For each field, bulker will make sure that a corresponding column exist in destination table. If not, Bulker will create it. Type will be best-guessed by value, or it could be explicitely set via type hint as in `{"a": "test", "__sql_type_a": "varchar(4)"}` * **Reliability**. Bulker will put the object to Kafka Queue immediately, so if datawarehouse is down, data won't be lost * **Streaming** or **Batching**. Bulker will send data to datawarehouse either as soon it become available in Kafka (streaming) or after some time (batching). Most data warehouses won't tolerate large number of inserts, that's why we implemented batching Bulker is a ๐Ÿ’œ of [Jitsu](https://github.com/jitsucom/jitsu), an open-source data integration platform. See full list of features below Bulker is also available as a go library if you want to embed it into your application as opposed to use a HTTP-server ## Features * ๐Ÿ›ข๏ธ **Batching** - Bulker sends data in batches in most efficient way for particular database. For example, for Postgres it uses COPY command, for BigQuery it uses batch-files * ๐Ÿšฟ **Streaming** - alternatively, Bulker can stream data to database. It is useful when number of records is low. Up to 10 records per second for most databases * ๐Ÿซ **Deduplication** - if configured, Bulker will deduplicate records by primary key * ๐Ÿ“‹ **Schema management** - Bulker creates tables and columns on the fly. It also flattens nested JSON-objects. Example if you send `{"a": {"b": 1}}` to bulker, it will make sure that there is a column `a_b` in the table (and will create it) * ๐Ÿฆพ **Implicit typing** - Bulker infers types of columns from JSON-data. * ๐Ÿ“Œ **Explicit typing** - Explicit types can be by type hints that are placed in JSON. Example: for event `{"a": "test", "__sql_type_a": "varchar(4)"}` Bulker will make sure that there is a column `a`, and it's type is `varchar(4)`. * ๐Ÿ“ˆ **Horizontal Scaling**. Bulker scales horizontally. Too much data? No problem, just add Bulker instances! * ๐Ÿ“ฆ **Dockerized** - Bulker is dockerized and can be deployed to any cloud provider and k8s. * โ˜๏ธ **Cloud Native** - each Bulker instance is stateless and is configured by only few environment variables. ## Supported databases Bulker supports the following databases: * โœ… PostgresSQL * โœ… Redshit * โœ… Snowflake * โœ… Clickhouse * โœ… BigQuery * โœ… MySQL * โœ… S3 * โœ… GCS Please see [Compatibility Matrix](.docs/db-feature-matrix.md) to learn what Bulker features are supported by each database. ## Documentation Links > **Note** > We highly recommend to read [Core Concepts](#core-concepts) below before diving into details * [How to use Bulker as HTTP Service](./.docs/server-config.md) * [Server Configuration](./.docs/server-config.md) * [HTTP API](./.docs/http-api.md) * How to use bulker as Go-lib *(coming soon)* ## Core Concepts ### Destinations Bulker operates with destinations. Destination is database or storage service (e.g. S3, GCS). Each destination has an ID and configuration which is represented by JSON object. Bulker exposes HTTP API to load data into destinations, where those destinations are referenced by their IDs. If destination is a database, you'll need to provide a destination table name. ### Event The main unit of data in Bulker is an *event*. Event is a represented JSON-object ### Batching and Streaming (aka Destination Mode) Bulker can send data to database in two ways: * **Streaming**. Bulker sends evens to destinaion one by one. It is useful when number of events is low (less than 10 events per second for most DBs). * **Batching**. Bulker accumulates events in batches and sends them periodically once batch is full or timeout is reached. Batching is more efficient for large amounts of events. Especially for cloud data-warehouses (e.g. Postgres, Clickhouse, BigQuery). ### Primary Keys and Deduplication Optionally, Bulker can deduplicate events by primary key. It is useful when you same event can be sent to Bulker multiple times. If available, Bulker uses primary keys, but for some data warehouses alternative strategies are used. >[Read more about deduplication ยป](./.docs/db-feature-matrix.md) --- ## File: connectors/airbytecdk/README.md # Airbyte - Golang SDK/CDK This package aims to help developers build connectors (sources/destinations) really fast in Go. The focus of this package is developer efficiency. It focusses on letting developers focus more on connector business logic instead of airbyte protocol knowledge. ## Installation ``` go get github.com/bitstrapped/airbyte ``` ## Docs - View the godoc's here: https://pkg.go.dev/github.com/bitstrapped/airbyte ## Usage ### By Example 1. The fastest way to get started it to look at the full example in `examples/httpsource` or the Example in the godoc ### Detailed Usage 1. Define a source by implementing the `Source` interface. ```go // Source is the only interface you need to define to create your source! type Source interface { // Spec returns the input "form" spec needed for your source Spec(logTracker LogTracker) (*ConnectorSpecification, error) // Check verifies the source - usually verify creds/connection etc. Check(srcCfgPath string, logTracker LogTracker) error // Discover returns the schema of the data you want to sync Discover(srcConfigPath string, logTracker LogTracker) (*Catalog, error) // Read will read the actual data from your source and use tracker.Record(), tracker.State() and tracker.Log() to sync data with airbyte/destinations // MessageTracker is thread-safe and so it is completely find to spin off goroutines to sync your data (just don't forget your waitgroups :)) // returning an error from this will cancel the sync and returning a nil from this will successfully end the sync Read(sourceCfgPath string, prevStatePath string, configuredCat *ConfiguredCatalog, tracker MessageTracker) error } ``` 2. Inside of main, pass your source into the sourcerunner ```go func main() { fsrc := filesource.NewFileSource("foobar.txt") runner := airbyte.NewSourceRunner(fsrc) err := runner.Start() if err != nil { log.Fatal(err) } } ``` 3. Write a dockerfile (sample below) ```dockerfile FROM golang:1.17-buster as build WORKDIR /base ADD . /base/ RUN go build -o /base/app . LABEL io.airbyte.version=0.0.1 LABEL io.airbyte.name=airbyte/source ENTRYPOINT ["/base/app"] ``` 4. Push to your docker repository and profit! ### Contributors - We'd like to give a shoutout and thank you to @ajzo90 and his initial work on https://github.com/ajzo90/airbyte-http-connector. @ajzo90's project inspired this project --- ## File: jitsubase/jsoniter/README.md [](https://sourcegraph.com/github.com/json-iterator/go?badge) [](https://pkg.go.dev/github.com/json-iterator/go) [](https://travis-ci.org/json-iterator/go) [](https://codecov.io/gh/json-iterator/go) [](https://goreportcard.com/report/github.com/json-iterator/go) [](https://raw.githubusercontent.com/json-iterator/go/master/LICENSE) [](https://gitter.im/json-iterator/Lobby) A high-performance 100% compatible drop-in replacement of "encoding/json" # Benchmark Source code: https://github.com/json-iterator/go-benchmark/blob/master/src/github.com/json-iterator/go-benchmark/benchmark_medium_payload_test.go Raw Result (easyjson requires static code generation) | | ns/op | allocation bytes | allocation times | | --------------- | ----------- | ---------------- | ---------------- | | std decode | 35510 ns/op | 1960 B/op | 99 allocs/op | | easyjson decode | 8499 ns/op | 160 B/op | 4 allocs/op | | jsoniter decode | 5623 ns/op | 160 B/op | 3 allocs/op | | std encode | 2213 ns/op | 712 B/op | 5 allocs/op | | easyjson encode | 883 ns/op | 576 B/op | 3 allocs/op | | jsoniter encode | 837 ns/op | 384 B/op | 4 allocs/op | Always benchmark with your own workload. The result depends heavily on the data input. # Usage 100% compatibility with standard lib Replace ```go import "encoding/json" json.Marshal(&data) ``` with ```go import jsoniter "github.com/json-iterator/go" var json = jsoniter.ConfigCompatibleWithStandardLibrary json.Marshal(&data) ``` Replace ```go import "encoding/json" json.Unmarshal(input, &data) ``` with ```go import jsoniter "github.com/json-iterator/go" var json = jsoniter.ConfigCompatibleWithStandardLibrary json.Unmarshal(input, &data) ``` [More documentation](http://jsoniter.com/migrate-from-go-std.html) # How to get ``` go get github.com/json-iterator/go ``` # Contribution Welcomed ! Contributors - [thockin](https://github.com/thockin) - [mattn](https://github.com/mattn) - [cch123](https://github.com/cch123) - [Oleg Shaldybin](https://github.com/olegshaldybin) - [Jason Toffaletti](https://github.com/toffaletti) Report issue or pull request, or email taowen@gmail.com, or [](https://gitter.im/json-iterator/Lobby) --- ## File: jitsubase/jsonorder/README.md [](https://sourcegraph.com/github.com/json-iterator/go?badge) [](https://pkg.go.dev/github.com/json-iterator/go) [](https://travis-ci.org/json-iterator/go) [](https://codecov.io/gh/json-iterator/go) [](https://goreportcard.com/report/github.com/json-iterator/go) [](https://raw.githubusercontent.com/json-iterator/go/master/LICENSE) [](https://gitter.im/json-iterator/Lobby) A high-performance 100% compatible drop-in replacement of "encoding/json" # Benchmark Source code: https://github.com/json-iterator/go-benchmark/blob/master/src/github.com/json-iterator/go-benchmark/benchmark_medium_payload_test.go Raw Result (easyjson requires static code generation) | | ns/op | allocation bytes | allocation times | | --------------- | ----------- | ---------------- | ---------------- | | std decode | 35510 ns/op | 1960 B/op | 99 allocs/op | | easyjson decode | 8499 ns/op | 160 B/op | 4 allocs/op | | jsoniter decode | 5623 ns/op | 160 B/op | 3 allocs/op | | std encode | 2213 ns/op | 712 B/op | 5 allocs/op | | easyjson encode | 883 ns/op | 576 B/op | 3 allocs/op | | jsoniter encode | 837 ns/op | 384 B/op | 4 allocs/op | Always benchmark with your own workload. The result depends heavily on the data input. # Usage 100% compatibility with standard lib Replace ```go import "encoding/json" json.Marshal(&data) ``` with ```go import jsoniter "github.com/json-iterator/go" var json = jsoniter.ConfigCompatibleWithStandardLibrary json.Marshal(&data) ``` Replace ```go import "encoding/json" json.Unmarshal(input, &data) ``` with ```go import jsoniter "github.com/json-iterator/go" var json = jsoniter.ConfigCompatibleWithStandardLibrary json.Unmarshal(input, &data) ``` [More documentation](http://jsoniter.com/migrate-from-go-std.html) # How to get ``` go get github.com/json-iterator/go ``` # Contribution Welcomed ! Contributors - [thockin](https://github.com/thockin) - [mattn](https://github.com/mattn) - [cch123](https://github.com/cch123) - [Oleg Shaldybin](https://github.com/olegshaldybin) - [Jason Toffaletti](https://github.com/toffaletti) Report issue or pull request, or email taowen@gmail.com, or [](https://gitter.im/json-iterator/Lobby) --- ## File: jitsubase/types/README.md # ๐Ÿ”ƒ github.com/elliotchance/orderedmap/v2 [](https://godoc.org/github.com/elliotchance/orderedmap/v2) ## Basic Usage An `*OrderedMap` is a high performance ordered map that maintains amortized O(1) for `Set`, `Get`, `Delete` and `Len`: ```go import "github.com/elliotchance/orderedmap/v2" func main() { m := orderedmap.NewOrderedMap[string, any]() m.Set("foo", "bar") m.Set("qux", 1.23) m.Set("123", true) m.Delete("qux") } ``` *Note: v2 requires Go v1.18 for generics.* If you need to support Go 1.17 or below, you can use v1. Internally an `*OrderedMap` uses the composite type [map](https://go.dev/blog/maps) combined with a trimmed down linked list to maintain the order. ## Iterating Be careful using `Keys()` as it will create a copy of all of the keys so it's only suitable for a small number of items: ```go for _, key := range m.Keys() { value, _:= m.Get(key) fmt.Println(key, value) } ``` For larger maps you should use `Front()` or `Back()` to iterate per element: ```go // Iterate through all elements from oldest to newest: for el := m.Front(); el != nil; el = el.Next() { fmt.Println(el.Key, el.Value) } // You can also use Back and Prev to iterate in reverse: for el := m.Back(); el != nil; el = el.Prev() { fmt.Println(el.Key, el.Value) } ``` The iterator is safe to use bidirectionally, and will return `nil` once it goes beyond the first or last item. If the map is changing while the iteration is in-flight it may produce unexpected behavior. --- ## File: sync-controller/examples/k8s/README.md # Example That is example of entities Sync Controller deploys to k8s cluster to run sync task. Maybe be outdated and in some details may not reflect current state of entities. But general idea should be the same. --- ## File: sync-controller/examples/docker-compose/README.md # Proof of concept This is proof of concept. Docker-compose runtime is not supported by Sync Controller --- ## File: sync-controller/README.md # Sync Controller Sync Controller is a component that is responsible for running sync tasks in kubernetes cluster. Sync Controller supports source connectors that use Airbyte protocol. All methods are supported: `spec`, `check`, `discover`, `read`. Sync Controller run all tasks in separate **Pods**. It runs pod and monitor its status, updating task status information in database tables. When task is finished, Sync Controller deletes Pod and frees resources. K8s **Secret** used to pass source credentials to Source connector in Pod. Each Pod contains of two main containers: Source connector and [Sync Sidecar](../sync-sidecar/README.md). Sync Sidecar captures data rows, logs, state and results of spec, discover and check command from Source connector: - Data rows are sent to the target destination in Bulker instance, - Logs are sent to preconfigured Bulkers destinations. - `spec`, `discover` and `check` results goes to Postgres database tables. ## Authorization All request should have `Authorization` header with value `Bearer ${token}`. Token should be one of the tokens defined in `SYNCCTL_AUTH_TOKENS` env variable, [see configuration section](#configuration). If `SYNCCTL_AUTH_TOKENS` is not configured, then no authorization is required. ## Endpoints ### `GET /spec` The endpoint is async. Run task to obtain Source connector specification. Resulting specification is stored in database table `source_spec`. Query parameters: - `package`* - source package image (for airbyte-based sources it's docker image) - `version`* - source package version - `connectorType` (optional) - `airbyte` by default Result table: `source_spec`. It writes a last JSON schema with update time by `[connectorType, package, version]`. It returns either `{"status": "ok"}` or `{"error": ...}` ### `POST /check` The endpoint is async. Run task to check Source connector credentials. Check result is stored in database table `source_check`. Query parameters: - `storageKey` - key to store results of check in a database - `package` - source package image (for airbyte-based sources it's docker image) - `version` - source package version Request body: ```json { "config": { // credentials json object } } ``` Result table: `source_check` ### `POST /discover` Run task to obtain Source connector streams catalog. Catalog is stored in database table `source_catalog`. Query parameters: - `storageKey` - key to store results of discover in a database - `package` - source package image (for airbyte-based sources it's docker image) - `version` - source package version Request body: ```json { "config": { // credentials json object } } ``` Result table: `source_catalog` ### `POST /read` Run task to pull data from Source connector. - Pulled data is sent to Bulker instance to `syncId` connection. - Task logs are inserted into `task_log` database table. - Task status is kept updated in `source_task` database table. - Saved state of task if any is stored in `source_state` database table. Query parameters: - `package` - source package image (for airbyte-based sources it's docker image) - `version` - source package version - `taskId` - id of this task. Should be unique for each task - `syncId` - id of sync entity (bulker destination id) where pulled events should be sent Request body: ```json { "config": { // credentials json object }, "catalog": { // catalog json object with selected streams and their configurations }, "state": { // json object with state from previous task run }, "destinationConfig": { // bulker destination credentials } } ``` ## Configuration | Environment variable | Required | Description | Default value | |------------------------------------------|----------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|-------------------------| | `SYNCCTL_DATABASE_URL` | yes | Postgres database URL. E.g.: `postgresql://user:pass@localhost:5432/postgres?sslmode=disable&search_path=bulker` | | | `SYNCCTL_HTTP_PORT` | no | HTTP port | `3043` | | `SYNCCTL_AUTH_TOKENS` | no | A list of auth tokens that authorizes user in HTTP interface separated by comma.Each token can be either:`${token}` un-encrypted token value`${salt}.${hash}` hashed token.`salt` should be random string`hash` = `hex(sha512(token + salt + SYNCCTL_TOKEN_SECRET))``token` may consist only of letters, digits, underscore and dash | | | `SYNCCTL_TOKEN_SECRET` | no | See above. A secret that is used for hashing tokens. | | | `SYNCCTL_SIDECAR_IMAGE` | yes | Sync Sidecar docker image. E.g.: `jitsucom/sidecar:latest` | | | | | `SYNCCTL_KUBERNETES_CLIENT_CONFIG` | yes | kubernetes client config in yaml format | | | `SYNCCTL_KUBERNETES_NAMESPACE` | no | kubernetes namespace where tasks pod should be run | `default` | | `SYNCCTL_CONTAINER_STATUS_CHECK_SECONDS` | no | frequency of running Pods status check | `10` | | `SYNCCTL_CONTAINER_INIT_TIMEOUT_SECONDS` | no | timeout for Pod initialization | `180` | | `SYNCCTL_SIDECAR_DATABASE_URL` | no | Must lead to the same db instance as `SYNCCTL_DATABASE_URL`. Use it when in k8s environment database is reachable through a different hostname or IP. | =`SYNCCTL_DATABASE_URL` | | `SYNCCTL_INSTANCE_ID` | no | ID of syncctl instance. It is used for metrics. If is not set, instance id will be generated and persisted to disk (`~/.syncctl/instance_id`) and reused on next restart. | random uuid | ## Database schema | Table name | Description | |------------------|------------------------------------------------------------------------------------------------------| | `source_spec` | Result of running `spec` command. Source connector specification that may be used to configure sync. | | `source_check` | Result of running `check` command. Source connector credentials check result. | | `source_catalog` | Result of running `discover` command. Source connector streams catalog. | | `source_task` | Status of running `read` command. | | `source_state` | State objects obtained while running `read` command. | | `task_log` | Logs of source tasks | --- ## File: sync-sidecar/README.md # ๐Ÿ๏ธ Sync Sidecar Sync Sidecar is meant to be used as sidecar container to Airbyte protocol compatible Source connectors in Kubernetes Pod or Docker Compose. Sync Sidecar captures data rows, logs, state and results of spec, discover and check command from Source connector: - Data rows are sent to the target destination in Bulker instance, - Logs are sent to preconfigured Bulkers destinations. - `spec`, `discover` and `check` results goes to Postgres database tables. ## Named Pipes Bulker-Sidecar uses named pipes to communicate with Source connector. In k8s environment all containers from the same Pod are running on the same machine, and it is possible to use named pipes. Source connectors entrypoint must be changed to direct their stderr and stdout output to named pipes. Volume with named pipes should be mounted to both Bulker-Sidecar and Source connector containers. InitContainer can be used to create named pipes in advance. ## Configuration - `STDOUT_PIPE_FILE` - path of named pipe for stdout of Source connector - `STDERR_PIPE_FILE` - path of named pipe for stderr of Source connector - `COMMAND` - Command that is used to run Source connector. Should be one of `spec`, `discover`, `check` or `read` - `STARTED_AT` - Timestamp when task was triggered - `DATABASE_URL` - URL of Postgres database where spec, discover, check results and read task statuses should be stored - `PACKAGE` - Name of Source connector package - `PACKAGE_VERSION` - Version of Source connector package - `STORAGE_KEY` - key to store results of `check` and `discover` commands in a database. - `SYNC_ID` - id of sync entity (bulker destination id) where pulled events should be sent. For `read` command. - `TASK_ID` - id of current running task ## 2. Official Technical Reference & Guides (jitsucom/docs) ## File: README.md # Introduction kSense helps businesses collect, correlate and perform action based on customer intentions. --- ## File: events-api/README.md # Events API ### Introduction kSense provides event API for exporting events to its analytics and prediction system. Events can have different meaning \(depending on purpose of client integration\). It could be web page views, mobile app usage, conversions, subscription / un-subscription events and so on. Moreover, user can define a custom event types. There're two ways of exporting events: through JavaScript call \(pixel\) and server-to-server. Both options have the same JSON API with minor differences in user-indentification process. All definitions are applicable to both JavaScript and JSON APIs. ### Basic concepts #### Event Type Each event has a type name. Some event types are pre-defined \(system even types\). Also, user can define its own type name. Each event may have a parent event type. It means that event can happen if only a parent type has happened before for the same user. System event types: | ID | Description | | :--- | :--- | | pv | Web Page View | | click | Click on a link | | conversion | Conversion: an final event in sales funnel. Such as purchase, form-submit and so on | | user\_info | An event for define a current user properties such as zip code.Each time user changes the | | subscribed | User subscribed to a service | | unsubscribed | User un-subscribed | | sub\_pause | Subscription pause | | sub\_resume | Subscription resume | | support\_contact | Contact support | Please see [Event JSON](event-json.md) page for further documentation. --- ## File: case-scenarios/README.md # Case Scenarios This section describes how event API should be used for different type of consumers: which event to send and what is the impact.