The Elasticsearch sink connector pulls messages from Pulsar topics and persists the messages to indexes.

How to get

This section describes how to build the Elasticsearch sink connector.

Work with Function Worker

You can get the NAR package of the Elasticsearch sink connector from the download page if you use Pulsar Function Worker to run connectors in a cluster.

Work with Function Mesh

You can pull the Elasticsearch sink connector Docker image from the Docker Hub if you use Function Mesh to run the connector.

How to configure

Before using the Elasticsearch sink, you need to configure it. This table lists the properties and the descriptions.

Work with Function Worker

You can create a configuration file (JSON or YAML) to set the properties if you use Pulsar Function Worker to run connectors in a cluster.

Examples for Elasticsearch 6.2+

Example

• JSON

  {
    "configs": {
      "elasticSearchUrl": "http://localhost:9200",
      "indexName": "my_index",
      "username": "scooby",
      "password": "doobie"
    }
  }
  

• YAML

  configs:
      elasticSearchUrl: "http://localhost:9200"
      indexName: "my_index"
      username: "scooby"
      password: "doobie"
  

Examples for Elasticsearch 6.2 and previous versions

• JSON

  {
    "elasticSearchUrl": "http://localhost:9200",
    "indexName": "my_index",
    "typeName": "doc",
    "username": "scooby",
    "password": "doobie"
  }
  

• YAML

  configs:
    elasticSearchUrl: "http://localhost:9200"
    indexName: "my_index"
    typeName: "doc"
    username: "scooby"
    password: "doobie"
  

Work with Function Mesh

You can create a CustomResourceDefinitions (CRD) to create a Elasticsearch sink connector. Using CRD makes Function Mesh naturally integrate with the Kubernetes ecosystem. For more information about Pulsar sink CRD configurations, see sink CRD configurations.

You can define a CRD file (YAML) to set the properties as below.

apiVersion: compute.functionmesh.io/v1alpha1
kind: Sink
metadata:
  name: es-sink-sample
spec:
  image: streamnative/pulsar-io-elastic-search:2.9.2.9
  className: org.apache.pulsar.io.elasticsearch.ElasticSearchSink
  replicas: 1
  maxReplicas: 1
  input:
    topics: 
    - persistent://public/default/destination
    typeClassName: “org.apache.pulsar.client.api.schema.GenericObject”
  sinkConfig:
    elasticSearchUrl: "http://localhost:9200"
    indexName: "my_index"
    typeName: "doc"
    username: "scooby"
    password: "doobie"
  pulsar:
    pulsarConfig: "test-pulsar-sink-config"
  resources:
    limits:
    cpu: "0.2"
    memory: 1.1G
    requests:
    cpu: "0.1"
    memory: 1G
  java:
    jar: connectors/pulsar-io-elastic-search-2.9.2.9.nar
  clusterName: test-pulsar
  autoAck: true

How to use

You can use the Elasticsearch sink connector with Function Worker or Function Mesh.

Work with Function Worker

You can make the Elasticsearch sink connector as a Pulsar built-in connector and use it on a standalone cluster or an on-premises cluster.

Standalone cluster

This example describes how to use the Elasticsearch sink connector to pull messages from Pulsar topics and persist the messages to indexes.

Prerequisites

• Install Docker (Community Edition).

Steps

1. Start a single-node Elasticsearch cluster.

   docker run -p 9200:9200 -p 9300:9300 \
    -e "discovery.type=single-node" \
    docker.elastic.co/elasticsearch/elasticsearch:7.13.3
   

2. Start Pulsar locally in standalone mode.

   bin/pulsar standalone
   

   Make sure that the NAR file is available at connectors/pulsar-io-elastic-search-{{pulsar:version}}.nar.

3. Start the Pulsar Elasticsearch connector in local run mode using one of the following methods.

   • Use the JSON configuration as shown previously.

     bin/pulsar-admin sinks localrun \
       --archive connectors/pulsar-io-elastic-search-{{pulsar:version}}.nar \
       --tenant public \
       --namespace default \
       --name elasticsearch-test-sink \
       --sink-config '{"elasticSearchUrl":"http://localhost:9200","indexName": "my_index","username": "scooby","password": "doobie"}' \
       --inputs elasticsearch_test
     

   • Use the YAML configuration file as shown previously.

     bin/pulsar-admin sinks localrun \
       --archive connectors/pulsar-io-elastic-search-{{pulsar:version}}.nar \
       --tenant public \
       --namespace default \
       --name elasticsearch-test-sink \
       --sink-config-file elasticsearch-sink.yml \
       --inputs elasticsearch_test
     

4. Publish records to the topic.

   bin/pulsar-client produce elasticsearch_test --messages "{\"a\":1}"
   

5. Check documents in Elasticsearch.

   • Refresh the index

     curl -s http://localhost:9200/my_index/_refresh
     

   • Search documents.

     curl -s http://localhost:9200/my_index/_search
     

     You can see the record that was published earlier has been successfully written into Elasticsearch.

     {"took":2,"timed_out":false,"_shards":{"total":1,"successful":1,"skipped":0,"failed":0},"hits":{"total":{"value":1,"relation":"eq"},"max_score":1.0,"hits":[{"_index":"my_index","_type":"_doc","_id":"FSxemm8BLjG_iC0EeTYJ","_score":1.0,"_source":{"a":1}}]}}
     

On-premises cluster

This example explains how to create a Elasticsearch sink connector in an on-premises cluster.

1. Copy the NAR package of the Elasticsearch connector to the Pulsar connectors directory.

   cp pulsar-io-Elasticsearch-{{connector:version}}.nar $PULSAR_HOME/connectors/pulsar-io-Elasticsearch-{{connector:version}}.nar
   

2. Reload all built-in connectors.

   PULSAR_HOME/bin/pulsar-admin sinks reload
   

3. Check whether the Elasticsearch sink connector is available on the list or not.

   PULSAR_HOME/bin/pulsar-admin sinks available-sinks
   

4. Create a Elasticsearch sink connector on a Pulsar cluster using the pulsar-admin sinks create command.

   PULSAR_HOME/bin/pulsar-admin sinks create \
   --sink-config-file <Elasticsearch-sink-config.yaml>
   

Work with Function Mesh

This example describes how to create a Elasticsearch sink connector for a Kuberbetes cluster using Function Mesh.

Prerequisites

• Create and connect to a Kubernetes cluster.
• Create a Pulsar cluster in the Kubernetes cluster.
• Install the Function Mesh Operator and CRD into the Kubernetes cluster.

Step

1. Define the Elasticsearch sink connector with a YAML file and save it as sink-sample.yaml.

   This example shows how to publish the Elasticsearch sink connector to Function Mesh with a Docker image.

   apiVersion: compute.functionmesh.io/v1alpha1
   kind: Sink
   metadata:
     name: es-sink-sample
   spec:
     image: streamnative/pulsar-io-elastic-search:2.9.2.9
     className: org.apache.pulsar.io.elasticsearch.ElasticSearchSink
     replicas: 1
     maxReplicas: 1
     input:
       topics: 
       - persistent://public/default/destination
       typeClassName: “org.apache.pulsar.client.api.schema.GenericObject”
     sinkConfig:
       elasticSearchUrl: "http://localhost:9200"
       indexName: "my_index"
       typeName: "doc"
       username: "scrappy"
       password: "fighter"
     pulsar:
       pulsarConfig: "test-pulsar-sink-config"
     resources:
       limits:
       cpu: "0.2"
       memory: 1.1G
       requests:
       cpu: "0.1"
       memory: 1G
     java:
       jar: connectors/pulsar-io-elastic-search-2.9.2.9.nar
     clusterName: test-pulsar
     autoAck: true
   

2. Apply the YAML file to create the Elasticsearch sink connector.

   Input

   kubectl apply -f <path-to-sink-sample.yaml>
   

   Output

   sink.compute.functionmesh.io/es-sink-sample created
   

3. Check whether the Elasticsearch sink connector is created successfully.

   Input

   kubectl get all
   

   Output

   NAME                                         READY   STATUS      RESTARTS   AGE
   pod/es-sink-sample-0               1/1    Running     0          77s
   

   After confirming the successful creation of the Elasticsearch sink connector, you can use it to export Pulsar messages to Elasticsearch.