Sql Sink

The sink will write the result to the database.

Compile & deploy plugin

This plugin must be used in conjunction with at least a database driver. We are using build tag to determine which driver will be included. This repository lists all the supported drivers.

This plugin supports sqlserver\postgres\mysql\sqlite3\oracle drivers by default. User can compile plugin that only support one driver by himself, for example, if he only wants mysql, then he can build with build tag mysql.

When using sqlserver as the target, you need to confirm that the sqlserver exposes port 1434.

Default build command

# cd $eKuiper_src
# go build -trimpath --buildmode=plugin -o plugins/sinks/Sql.so extensions/sinks/sql/sql.go
# cp plugins/sinks/Sql.so $eKuiper_install/plugins/sinks

MySql build command

# cd $eKuiper_src
# go build -trimpath --buildmode=plugin -tags mysql -o plugins/sinks/Sql.so extensions/sinks/sql/sql.go
# cp plugins/sinks/Sql.so $eKuiper_install/plugins/sinks

Properties

Property name	Optional	Description
url	false	The url of the target database
table	false	The table name of the result
fields	true	The fields to be inserted to. The result map and the database should both have these fields. If not specified, all fields in the result map will be inserted.
tableDataField	true	Write the nested values of the tableDataField into database.
rowkindField	true	Specify which field represents the action like insert or update. If not specified, all rows are default to insert.

Other common sink properties are supported. Please refer to the sink common properties for more information.

You can check the connectivity of the corresponding sink endpoint in advance through the API: Connectivity Check

Dynamic field names

When fields is not configured, the SQL sink derives column names from the result map keys (from the first row for a batch). Each derived name must match [A-Za-z_][A-Za-z0-9_]*: it must start with an ASCII letter or underscore and may then contain only ASCII letters, digits, or underscores. This restriction also applies to keys named by rowkindField when they are included in the generated columns.

If a derived name does not match this format, the affected write is rejected before SQL is executed. The sink does not silently drop or automatically quote the invalid key.

Explicitly configured table, fields, and keyField values are passed to the generated SQL unchanged so that database-specific identifier syntax remains supported. Each configured fields entry is also used to look up the value in the result map, so the map key must exactly match the configured entry and the entry must use syntax accepted by the target database.

Sample usage

Below is a sample for using sql to get the target data and set to mysql database

{
  "id": "rule",
  "sql": "SELECT stuno as id, stuName as name, format_time(entry_data,\"YYYY-MM-dd HH:mm:ss\") as registerTime FROM SqlServerStream",
  "actions": [
    {
      "log": {
      },
      "sql": {
        "url": "mysql://user:test@140.210.204.147/user?parseTime=true",
        "table": "test",
        "fields": ["id","name","registerTime"]
      }
    }
  ]
}

Write values of tableDataField into database:

The following configuration will write telemetry field's values into database

{
  "telemetry": [{
    "temperature": 32.32,
    "humidity": 80.8,
    "ts": 1388082430
  },{
    "temperature": 34.32,
    "humidity": 81.8,
    "ts": 1388082440
  }]
}

{
  "id": "rule",
  "sql": "SELECT telemetry FROM dataStream",
  "actions": [
    {
      "log": {
      },
      "sql": {
        "url": "mysql://user:test@140.210.204.147/user?parseTime=true",
        "table": "test",
        "fields": ["temperature","humidity"],
        "tableDataField":  "telemetry",
      }
    }
  ]
}

Update Sample

By specifying the rowkindField and keyField, the sink can generate insert, update or delete statement against the primary key.

{
  "id": "ruleUpdateAlert",
  "sql":"SELECT * FROM alertStream",
  "actions":[
    {
      "sql": {
        "url": "sqlite://test.db",
        "keyField": "id",
        "rowkindField": "action",
        "table": "alertTable",
        "sendSingle": true
      }
    }
  ]
}