Back

Integrate the ElasticSearch Data Pipeline Connector

Learn how to integrate the ElasticSearch Data Pipeline Connector.

This guide covers:

Get the Source Data
Register a data pipe and send source data to configured target ElasticSearch database
Verify the target database received the data

1. Prerequisites

To complete this guide, you need:

Roughly 15 minutes
An IDE
JDK 11+ installed with JAVA_HOME configured appropriately
Apache Maven 3.9.5

Verify Maven is using the Java you expect

If you have multiple JDK’s installed, it is not certain Maven will pick up the expected java and you could end up with unexpected results. You can verify which JDK Maven uses by running mvn --version.

Download the Braineous-1.0.0-CR3 zip archive

This tutorial is located under: braineous-1.0.0-cr3/tutorials/elastic-connector

2. Initialize

Get an instance of the DataPlatformService. Setup your API_KEY and API_SECRET. Please refer to 'Step 9' of the Getting Started guide. https://bugsbunnyshah.github.io/braineous/get-started/

DataPlatformService dataPlatformService = DataPlatformService.getInstance();

String apiKey = "ffb2969c-5182-454f-9a0b-f3f2fb0ebf75";
String apiSecret = "5960253b-6645-41bf-b520-eede5754196e";

3. Get the Source Data

Let’s start with a simple Json array to be used as datasource to be ingested by the Braineous Data Ingestion Engine

Source Data

[
  {
    "id" : 1,
    "name": "name_1",
    "age": 46,
    "addr": {
      "email": "name_1@email.com",
      "phone": "123"
    }
  },
  {
    "id": "2",
    "name": "name_2",
    "age": 55,
    "addr": {
      "email": "name_2@email.com",
      "phone": "1234"
    }
  }
]

Java Code

String datasetLocation = "dataset/data.json";
String json = Util.loadResource(datasetLocation);

A dataset can be loaded from any data source such as a database, legacy production data store, live data feed, third-party data source, Kafka stream, etc. In this example the dataset is loaded from a classpath resource located at src/main/resources/dataset/data.json

4. Register a data pipe and send source data to configured target ClickHouse database

Register a data pipe with the Braineous Data Ingestion Engine using the Java Braineous Data Ingestion Client SDK.

Pipe Configuration

{
  "pipeId": "yyya",
  "entity": "abc",
  "configuration": [
    {
      "stagingStore" : "com.appgallabs.dataplatform.targetSystem.core.driver.ElasticSearchStagingStore",
      "name": "yyya",
      "config": {
        "elasticSearchUrl": "http://localhost:9200/",
        "index": "yyya",
        "security": {
          "type": "built_in_users",
          "user": "elastic",
          "password": "password"
        },
        "jsonpathExpressions": []
      }
    }
  ]
}

pipeId : As a data source provider, this id identifies this data pipe uniquely with the Braineous Data Pipline Engine.
entity : The business/domain entity that this dataset should be associated with.
configuration.stagingStore: The Staging Store driver
configuration.name: a user-friendly way to indentify the target store
configuration.config.elasticSearchUrl: Elastic Search URl of your ElasticSearch instance
configuration.config.index: Elastic Search Index for the Staging Store
configuration.config.security.type: Elastic Search authentication type
configuration.config.security.user: Elastic Search user
configuration.config.security.password: Elastic Search user’s password
configuration.jsonpathExpressions: Data Transformation based on JSONPath specification: https://www.ietf.org/archive/id/draft-goessner-dispatch-jsonpath-00.html

A data pipe can be configured with multiple target stores/systems associated with the same data pipe for data delivery.

Java Code - Register Pipe

String configLocation = "pipe_config/pipe_config.json";
String pipeConfigJson = Util.loadResource(configLocation);
JsonObject configJson = JsonUtil.validateJson(pipeConfigJson).getAsJsonObject();
String pipeId = configJson.get("pipeId").getAsString();
String entity = configJson.get("entity").getAsString();
System.out.println("*****PIPE_CONFIGURATION******");
JsonUtil.printStdOut(configJson);

//configure the DataPipeline Client
Configuration configuration = new Configuration().
ingestionHostUrl("http://localhost:8080/").
apiKey(apiKey).
apiSecret(apiSecret).
streamSizeInObjects(0);
dataPlatformService.configure(configuration);

//register pipe
dataPlatformService.registerPipe(configJson);
System.out.println("*****PIPE_REGISTRATION_SUCCESS******");

Pipe Configuration can be provided dynamically at runtime. The source can be a database, a configuration system, local file system, network file system etc. In this example the dataset is loaded from a classpath resource located at src/main/resources/pipe_config/pipe_config.json

Java Code - Send Data for ingestion

//send source data through the pipeline
dataPlatformService.sendData(pipeId, entity,datasetElement.toString());
System.out.println("*****DATA_INGESTION_SUCCESS******");

5. Run the Tutorial

cd braineous-1.0.0-cr3/tutorials/clickhouse-connector

./run.sh

Expected Output:

SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
*****DATA_SET******
******ARRAY_SIZE: 4**********
[
  {
    "id": 1,
    "name": "name_1",
    "age": 46,
    "addr": {
      "email": "name_1@email.com",
      "phone": "123"
    }
  },
  {
    "id": 2,
    "name": "name_2",
    "age": 55,
    "addr": {
      "email": "name_2@email.com",
      "phone": "1234"
    }
  },
  {
    "id": 3,
    "name": "name_3",
    "age": 60,
    "addr": {
      "email": "name_3@email.com",
      "phone": "12345"
    }
  },
  {
    "id": 4,
    "name": "name_4",
    "age": 65,
    "addr": {
      "email": "name_4@email.com",
      "phone": "123456"
    }
  }
]
**********************
*****PIPE_CONFIGURATION******
{
  "pipeId": "yyya",
  "entity": "abc",
  "configuration": [
    {
      "stagingStore": "com.appgallabs.dataplatform.targetSystem.core.driver.ElasticSearchStagingStore",
      "name": "yyya",
      "config": {
        "elasticSearchUrl": "http://localhost:9200/",
        "index": "yyya",
        "security": {
          "type": "built_in_users",
          "user": "elastic",
          "password": "password"
        },
        "jsonpathExpressions": []
      }
    }
  ]
}
**********************
*****PIPE_REGISTRATION_SUCCESS******
***SENDING_DATA_START*****
*****DATA_INGESTION_SUCCESS******

6. Verify all the data is successfully stored in the Elastic Search Database

To verify the success of the ingestion and delivery to the configured target databases, refer to the Braineous Server Console.

Expected Result :

2024-07-25 20:50:18,663 INFO  [org.apa.fli.tab.cat.hiv.HiveCatalog] (pool-10-thread-3) Setting hive conf dir as /Users/babyboy/mumma/braineous/infrastructure/apache-hive-3.1.3-bin/conf
{
  "status_code": 200,
  "response": {
    "errors": false,
    "took": 598,
    "items": [
      {
        "index": {
          "_index": "yyya",
          "_id": "oR-E7JABE6xho924ZSQe",
          "_version": 1,
          "result": "created",
          "_shards": {
            "total": 2,
            "successful": 1,
            "failed": 0
          },
          "_seq_no": 33,
          "_primary_term": 1,
          "status": 201
        }
      },
      {
        "index": {
          "_index": "yyya",
          "_id": "oh-E7JABE6xho924ZSQg",
          "_version": 1,
          "result": "created",
          "_shards": {
            "total": 2,
            "successful": 1,
            "failed": 0
          },
          "_seq_no": 34,
          "_primary_term": 1,
          "status": 201
        }
      },
      {
        "index": {
          "_index": "yyya",
          "_id": "ox-E7JABE6xho924ZSQg",
          "_version": 1,
          "result": "created",
          "_shards": {
            "total": 2,
            "successful": 1,
            "failed": 0
          },
          "_seq_no": 35,
          "_primary_term": 1,
          "status": 201
        }
      },
      {
        "index": {
          "_index": "yyya",
          "_id": "pB-E7JABE6xho924ZSQg",
          "_version": 1,
          "result": "created",
          "_shards": {
            "total": 2,
            "successful": 1,
            "failed": 0
          },
          "_seq_no": 36,
          "_primary_term": 1,
          "status": 201
        }
      }
    ]
  }
}
**********************

1. Prerequisites
2. Initialize
3. Get the Source Data
4. Register a data pipe and send source data to configured target ClickHouse database
5. Run the Tutorial
6. Verify all the data is successfully stored in the Elastic Search Database