Ongoing Tasks: Elasticsearch ETL

• The following steps are required when creating an Elasticsearch ETL task:

  • Define a connection string which includes:
    • URLs to Elasticsearch nodes.
    • Authentication method required by the Elasticsearch nodes.
  • Define the Elasticsearch Indexes
    • Indexes are used by Elasticsearch to store and locate documents.
    • The ETL task will send new documents to the specified Elasticsearch indexes.
    • If not otherwise specified, existing Elasticsearch documents will be removed before adding new documents.
    • A document identifier field property is defined per document, and used by the delete command to locate the matching documents.
  • Define Transformation Scripts.
    The transformation script determines which RavenDB documents will be transferred, to which Elasticsearch Indexes, and in what form.

• For a thorough step-by-step explanation:

  • Learn here to define an Elasticsearch ETL task using code.
  • Learn here to define an Elasticsearch ETL task using Studio.

• The structure and syntax of an Elasticsearch ETL transformation script are similar to those of all other ETL types (RavenDB ETL, SQL ETL, and OLAP ETL) scripts.
  The script defines which documents will be Extracted from the database, Transforms the retrieved data, and Loads it to the Elasticsearch destination. Learn about ETL transformation scripts here.

• The script Loads data to the Elasticsearch destination using the loadTo<Target>(obj) command.

  • Target is the name of the Elasticsearch index to which the data is transferred.
    • In the task settings:
      Define Elasticsearch Index names using only lower-case characters (as required by Elasticsearch).
      E.g. orders
    • In the transformation script:
      The target can be defined using both upper and lower-case characters.
      The task will transform the index name to all lower-case characters before sending it to Elasticsearch.
      E.g. use either loadToOrders or loadToorders.
  • obj is an object defined by the script, that will be loaded to Elasticsearch.
    It determines the shape and contents of the document that will be created on the Elasticsearch Index.
    E.g., the following script defines the orderData object and loads it to the orders index:
    

    var orderData = { DocId: id(this),
                      OrderLinesCount: this.Lines.length,
                      TotalCost: 0 };
    
    loadToOrders(orderData);

What is Transferred

An Elasticsearch ETL task transfers documents only.
Document extensions like attachments, counters, or time series, will not be transferred.

Document Identifiers

• When Elasticsearch stores RavenDB documents, it provides each of them with an automatically-generated iD.
• RavenDB needs to delete and replace documents, but cannot do it using Elasticsearch's arbitrarily generated IDs.
  Instead, one of the transferred document's properties is used as ID.
• The identifier must be a property that the transformation script passes to Elasticsearch.
  To achieve this:
  • Add a dedicated property to the transferred data structure in your script, that will hold the original RavenDB document ID.
    The property's Name can be any name of your choice.
    The property's Value must be: id(this)
  • E.g., the DocId property below is used to hold the RavenDB document ID in the transferred document.
    

    var orderData = {
                     DocId: id(this), // document ID property
                     OrderLinesCount: this.Lines.length,
                     TotalCost: 0
                      };
    
    loadToOrders(orderData);

• In addition to specifying this document property in the script, it must be defined for the ETL task:
  • Either set DocumentIdProperty through code (see code sample),
  • or Set the Document ID Property Name field via Studio.

Transactions

The task delivers the data to the Elasticsearch destination in one or two calls per index.

1. _delete_by_query:
   An optional command, to delete existing versions of RavenDB documents from Elasticsearch before appending new ones.
   

   POST orders/_delete_by_query?refresh=true
   {"query":{"terms":{"DocID":["orders/1-a"]}}}

2. _bulk :
   Append RavenDB documents to the Elasticsearch destination.
   

   POST orders/_bulk?refresh=wait_for
   {"index":{"_id":null}}
   {"OrderLinesCount":3,"TotalCost":0,"DocID":"orders/1-a"}

Insert Only Mode

You can enable the task's Insert Only mode using code or via Studio, to omit _delete_by_query commands and so refrain from deleting documents before the transfer.

• When the Elasticsearch ETL task runs for the very first time, it will create any Elsasticsearch index defined in the task that doesn't exist yet.

• When the index is created, the document property that holds the RavenDB document ID will be defined as a non-analyzed field, with type keyword to avoid having full-text-search on it.
  This way the RavenDB document identifiers won't be analyzed and the task will be able to _delete_by_query using exact match on those IDs.
  I.e.
  

  PUT /newIndexName
  {
    "mappings": {
        "properties": {
            "DocId": {   // the DocumentIdProperty
                "type": "keyword"
            }
        }
     }
  }

Add an Elasticsearch ETL Task

• To define an Elasticsearch ETL task through the client, use the AddEtlOperation API method as shown below.
  Pass it an ElasticSearchEtlConfigurationinstance with -
  • The name of a defined Connection String.
    You can define a connection string using code or via Studio.
  • A list of Elasticsearch Indexes.
  • A list of Transformation Scripts.

Code Sample:

// Create an Elasticsearch ETL task
AddEtlOperation<ElasticSearchConnectionString> operation = new AddEtlOperation<ElasticSearchConnectionString>(
new ElasticSearchEtlConfiguration()
{
    ConnectionStringName = elasticSearchConnectionString.Name, // Connection String name
    Name = "ElasticsearchEtlTask", // ETL Task name
        
    ElasticIndexes =
    {
        // Define Elasticsearch Indexes
        new ElasticSearchIndex { // Elasticsearch Index name
                                 IndexName = "orders", 
                                 // The Elasticsearch document property that will contain
                                 // the source RavenDB document id.
                                 // Make sure this property is also defined inside the
                                 // transform script.
                                 DocumentIdProperty = "DocId", 
                                 InsertOnlyMode = false }, 
        new ElasticSearchIndex { IndexName = "lines",
                                 DocumentIdProperty = "OrderLinesCount", 
                                 // If true, don't send _delete_by_query before appending docs
                                 InsertOnlyMode = true 
                               }
    },
    Transforms =
    {   // Transformation script configuration
        new Transformation()
        {
            // RavenDB collections that the script uses
            Collections = { "Orders" }, 

            Script = @"var orderData = {
                       DocId: id(this),
                       OrderLinesCount: this.Lines.length,
                       TotalCost: 0
                       };

                       // Write the `orderData` as a document to the Elasticsearch 'orders' index
                       loadToOrders(orderData);", 
            
            // Transformation script Name
            Name = "TransformIDsAndLinesCount" 
        }
    }
});

store.Maintenance.Send(operation);

Add an Elasticsearch Connection String

• An Elasticsearch connection string includes a list of Elasticsearch destinations URLs, and determines the Authentication Method required to access them.
  • Omit the Authentication property if the Elasticsearch destination requires no authentication.
  • Add a connection string as shown below.

Code Sample:

// Create a Connection String to Elasticsearch
var elasticSearchConnectionString = new ElasticSearchConnectionString
{
    // Connection String Name
    Name = "ElasticConStr", 
    // Elasticsearch Nodes URLs
    Nodes = new[] { "http://localhost:9200" }, 
    // Authentication Method
    Authentication = new Raven.Client.Documents.Operations.ETL.ElasticSearch.Authentication 
    { 
        Basic = new BasicAuthentication
        {
            Username = "John",
            Password = "32n4j5kp8"
        }
    }
};

store.Maintenance.Send(new PutConnectionStringOperation<ElasticSearchConnectionString>(elasticSearchConnectionString));