roadiejs-import

A plugin for RoadieJS

A configurable ETL pipeline, based on Node.js streams.

Contents

• API
  • createImportStream
  • getImportStreamStatus
  • getImportStreamMessages
• Elements
  • importStream
  • csvRecord
  • populate
• License

API

createImportStream

Registers a new import stream, and goes on to start importing.

Request

POST /streams

{
  "namespace": "roadietest",
  "blueprintName": "importPlanets",
  "blueprintVersion": 1,
  "localVersion": 0,
  "importStream": "planetCsv",
  "missingAction": "error",
  "source": {
    "type": "file",
    "options": {
      "paths": "./test/planets/import_files/advanced.csv"
    }
  }
}

| Name | Notes | ---- | ---- | namespace | Namespace of the blueprint which contains the import element. | blueprintName | Name of the blueprint which contains the import element. | blueprintVersion | Version number of the blueprint which contains the import element. | localVersion | Local version number of the blueprint which contains the import element. | importStream | The id of an importStream element, that's defined in the identified blueprint. | missingAction | Controls the behaviour if trying to update/delete a document that doesn't exist. Valid values are warning, error. | source | An object to configure the source of the import.

Sources

Data can be streamed from multiple sources. The source object is therefore mandatory, and has two common keys:

| Name | Notes | ---- | ---- | type | The type of import (e.g. file). | options | An object containing config specific to the type of import (see below).

file

Imports data from files stored on the local file system.

"source": {
  "type": "file",
  "options": {
    "paths": "./test/school/import_files/full_school_dump.csv"
  }
}

| Name | Notes | ---- | ---- | paths | Defines which files should be loaded. Supports paths to single files, * wildcards, glob-style ** (for directory recursion) and arrays of multiple strings.

Response

Status 201

{
  "_id": "557c3834f0f1c14e25220e8b",
  "_created": "2015-06-13T14:03:32.571Z",
  "namespace": "roadietest",
  "blueprintName": "importPlanets",
  "blueprintVersion": 1,
  "blueprintLocalVersion": 0,
  "importStream": "planetCsv",
  "totalSize": 1147,
  "status": "starting",
  "processedSize": 0,
  "count": 0,
  "warnings": 0,
  "failures": 0,
  "notDealtWith": 0
}

| Name | Notes | ---- | ---- | _id | The unique database-generated id for the import process. | _created | A timestamp of when the import was registered. | namespace | Namespace of the blueprint, as supplied in the request. | blueprintName | Name of the blueprint, as supplied in the request. | blueprintVersion | Version of the blueprint, as supplied in the request. | blueprintLocalVersion | Local version of the blueprint, as supplied in the request. | importStream | The id of an importStream element, as supplied in the request. | totalSize | The number of units the import is estimated to be. Most likely number of bytes. | status | Current status of the import (expect starting). | processedSize | How many units have been imported so far (expect 0 at this point). | count | Total count of documents which have been processed (regardless of whether they succeeded or failed). | warnings | Number of documents, within the overall count, that have raised a warning. | failures | Number of documents, within the overall count, that have failed. | notDealtWith | Number of documents, within the overall count, that did not match any record pattern.

getImportStreamStatus

Get the latest status of a flow.

Request

GET /streams/:id

| Name | Notes | ---- | ---- | id | The id that uniquely identifies an import (e.g. the _id value returned from createImportStream).

Response

Status 200

{
  "_id": "557c6f1487a62fff374fa2ed",
  "_created": "2015-06-13T17:57:40.707Z",
  "namespace": "roadietest",
  "blueprintName": "importSchools",
  "blueprintVersion": 1,
  "blueprintLocalVersion": 0,
  "importStream": "studentCsv",
  "totalSize": 364,
  "status": "succeeded",
  "finished": "2015-06-13T17:57:40.820Z",
  "processedSize": 364,
  "count": 6,
  "warnings": 0,
  "failures": 0,
}

| Name | Notes | ---- | ---- | _id | The unique database-generated id for the import process (e.g. the id provided as a parameter as part of the request). | _created | A timestamp of when the import was registered. | namespace | Namespace of the blueprint, as supplied in the request. | blueprintName | Name of the blueprint, as supplied in the request. | blueprintVersion | Version of the blueprint, as supplied in the request. | blueprintLocalVersion | Local version of the blueprint, as supplied in the request. | importStream | The id of an importStream element, as supplied in the request. | totalSize | The number of units the import is estimated to be. Most likely number of bytes. | status | Current status of the import, valid values are starting, succeeding, warning, failing, warned, failed, succeeded. | finished | Timestamp of when the flow finished (not present if it's still running). | processedSize | How many units have been imported so far | count | Total count of documents which have been processed (regardless of whether they succeeded or failed). | warnings | Number of documents, within the overall count, that have raised a warning. | failures | Number of documents, within the overall count, that have failed. | notDealtWith | Number of documents, within the overall count, that did not match any record pattern.

getImportStreamMessages

Returns an array of messages that have been generated by the specified flow (ordered-by creation timestamp ascending).

Request

GET /streams/:id/messages

| Name | Notes | ---- | ---- | id | The id that uniquely identifies an import (e.g. the _id value returned from createImportStream).

Response

Status 200

[
  { "_id": "557c7680f09749d93b88619f",
    "transactionId": "557c7680f09749d93b88619a",
    "schemaName": "students",
    "namespace": "roadietest",
    "blueprintName": "importSchools",
    "blueprintVersion": 1,
    "blueprintLocalVersion": 0,
    "type": "warning",
    "name": "noDoc",
    "message": "Unable to find document"
  }
]

| Name | Notes | ---- | ---- | _id | A unique value to identify the message. | transactionId | The unique database-generated id for the import process (e.g. the id provided as a parameter as part of the request). | schemaName | The id of a schema related to the message. | namespace | Namespace of the blueprint responsible for the import. | blueprintName | Name of the blueprint responsible for the import. | blueprintVersion | Version of the blueprint responsible for the import. | blueprintLocalVersion | Local version of the blueprint responsible for the import. | type | Type of message: a value from info, warning, error or exception. | name | Name (e.g. code) of the message. | message | Short message content | body | Data to support the message (content specific to the type/name of message)

Elements

importStream

Registers a new import (e.g. a way of importing data into schemas within the blueprint).

Example

{
  "id": "planetCsv",
  "element": "importStream",
  "config": {
    "parser": {
      "type": "csv",
      "options": {
        "delimiter": ",",
        "qualifier": "\\"
      }
    },
    "target": {
      "type": "data"
    }
  }
}

Config

| Name | Type | Notes | ----------- | -------| ----------- | parser | object | An object that should contain a type string (e.g. csv) for identifying a parser, and an options object for configuring the parser. | target| object | An object that configures a supported target for the import. The object must include a type value to identify a target.

Parsers

A parser takes the raw data stream from a source (configured via createImportStream) and turns it into a usable object for passing onto an adaptor.

csv

The csv parser expects a source that can provide individual chunks of data (typically a line from a file).

• Internally, parsing is handled via the csv-parse package.
• The options defined for the parser are passed through to a csv-parse parser. More information here.

Adaptors

An adaptor takes the output of a parser and maps it to fields in a schema.

• There's no need to explicitly define an adaptor.
• If an adaptor hasn't been defined, then an adaptor with the same name as the parser is used.
• The behaviour of an adaptor depends on its type.

csv

The csv adaptor expects one or more csvRecord elements to be defined as a child element of the importStream element.

Targets

A target is the final destination in the import pipeline, and does something with the output of an adaptor.

data

• Hooks into roadiejs-data so the object produced out of the adaptor can persisted.

console

• Outputs the object produced out of the adaptor to the console.

csvRecord

If an importStream element has a parser of type csv, then one or more csvRecord child elements should be defined for it.

• The purpose of a csvRecord is to transform the output of a csv parser to a schema/field structure.
• Multiple csvRecord elements can be configured under an importStream element - as it's possible to 'identify' a suitable schema from the available csv columns.
• A special csv array will be accessible when evaluating expressions, this reflects the parsed columns from the underlying CSV data.

Example

{
  "id": "craterRecord",
  "element": "csvRecord",
  "parent": "importStream.planetCsv",
  "config": {
    "schemaId": "planets",
    "recordIdentification": "csv[0]=='crater'",
    "actionIdentification": {
      "post": "csv[1]=='I'",
      "put": "csv[1]=='U'",
      "upsert": "csv[1]=='M'",
      "del": "csv[1]=='D'"
    },
    "paramMap": [
      "csv[5]",
      "moons",
      "csv[6]",
      "craters",
      "csv[2]"
    ],
    "data": {
      "title": "csv[3]",
      "diameter": "csv[4]"
    }
  }
}

Config

| Name | Type | Notes | ------------ | -------- | ----------- | schemaId | string | The id of a schema defined within the blueprint that the CSV data will be ultimately persisted.. | recordIdentification| string | Optional. An expression. If it evaluates to true then the config of this csvRecord element will be used to transform the CSV data into a field structure. | actionIdentification| object | Optional. Maps an action (e.g. post, put, upsert or del) to an expression. If it evaluates to true then that action will be used to persist/delete the transformed data. | paramMap | [String] | Optional. An array of strings. Maps parameters (starting at docId) of a /data route to the contents of the CSV record. It is therefore possible target sub-docs. | data | object | Maps a field name to an expression. The result of the expression will then be used as the value for that field.

populate

A simple way to populate with data - useful for supplying reference/lookup data from within a blueprint definition.

• Ensure a populate element is a child of the a schema element you wish to populate.
• Schemas will only ever be populated once, and will not be re-asserted every time the blueprint is used

Example

{
  "id": "statesPopulator",
  "element": "populate",
  "parent": "schema.states",
  "config": {
    "map": [
      "name",
      "abbreviation",
      "capitalCity",
      "mostPopulatedCity",
      "population",
      "squareMiles"
    ],
    "data": [
      ["ALABAMA", "AL", "Montgomery", "Birmingham", 4708708, 52423],
      ["ALASKA", "AK", "Juneau", "Anchorage", 698473, 656425],
      ["ARIZONA", "AZ", "Phoenix", "Phoenix", 6595778, 114006],
      ["ARKANSAS", "AR", "Little Rock", "Little Rock", 2889450, 53182],
      ["CALIFORNIA", "CA", "Sacramento", "Los Angeles", 36961664, 163707],
      ["COLORADO", "CO", "Denver", "Denver", 5024748, 104100],
      ["CONNECTICUT", "CT", "Hartford", "Bridgeport", 3518288, 5544]
    ]
  }
}

Config

| Name | Type | Notes | ------------ | -------| ----------- | map | [string] | An array of strings, each a field name within the schema you wish to populate. The order is important... | data| [array] | An array of arrays - mimicking a record/field structure. The values of each 'record' should be in the same order as defined in map.

License

MIT