ElasticSearch source design documention

Background and Motivation

  • the connector of flink with ES just support sink but doesn’t support source, some company truly has the scenario of reading elasticsearch
  • We propose a throughout es source connector implementation in Flink.

Scope

  • We’ll design and develop a es source connector with the following features:
    • Es as bounded source
    • support multiple es version(5/6/7)
    • support FilterPushDown and ProjectPushDown and LimitPushDown to optimize query performance

Overall Design

  • We split logical data into xxInputSplits in hadoop ecosystem. and use xxInputFormt to read data
  • implementation DynamicTableSource to support sql

ElasticsearchInputSplit

  • we split es type(it is index in high version, because every index just has one default type, and default type is docs) into different ElasticSearchInputSplit. One split corresponds to one shard.
    EsSource design doc - 图1

ElasticsearchRowDataInputformat

public class ElasticSearchRowFormat extends RichInputFormat {

private static final Logger LOG = LoggerFactory.getLogger(ElasticSearchRowDataInputFormat.class);

//for parser json data

private DeserializationSchema serializationSchema;

// for FilterPushDown

private QueryBuilder predict;

private String index;

private SearchHit[] currentScrollWindowHits = null;

private int nextRecordIndex = 0;

private transient long currentReadCount = 0L;

// fir LimitPushDown to limit data to fecth

private long limit;

private final int scrollTimeout;

private final int scrollSize;

// ————————————————————————————————————

// User-facing API and configuration

// ————————————————————————————————————

/**

  • The config map that contains configuration for the bulk flushing behaviours.

    *

  • For {@link org.elasticsearch.client.transport.TransportClient} based implementations, this config

  • map would also contain Elasticsearch-shipped configuration, and therefore this config map

  • would also be forwarded when creating the Elasticsearch client.

    */

private final Map userConfig;

// ————————————————————————————————————

// Internals for the Flink Elasticsearch Sink

// ————————————————————————————————————

/* Call bridge for different version-specific. /

private final ElasticsearchApiCallBridge callBridge;

/* Elasticsearch client created using the call bridge. /

private transient C client;

public ElasticSearchRowInputFormat(

  1.  ElasticsearchApiCallBridge<C> callBridge,
  3.  Map<String, String> userConfig,
  5.  DeserializationSchema<RowData> serializationSchema,
  7.  QueryBuilder queryBuilder,
  9.  String index,
  11.  long limit) {
  13.  this.serializationSchema = serializationSchema;
  15.  this.queryBuilder = queryBuilder;
  17.  this.index = index;
  19.  this.limit = limit;
  21.  this.callBridge = checkNotNull(callBridge);
  25.  checkNotNull(userConfig);
  29.  // copy config so we can remove entries without side-effects
  31.  this.userConfig = new HashMap<>(userConfig);
  35. //todo

}

@Override

public void configure(Configuration parameters) {

  1.  client = callBridge.createClient(userConfig);
  3.  callBridge.verifyClientConnection(client);
  5. //todo

}

@Override

public void open(ElasticsearchInputSplit split) throws IOException {

  1. LOG.info("Opening split: {}", split);
  5. SearchRequestBuilder searchRequestBuilder = client.prepareSearch(index)
  7.         .setTypes(type)
  9.         .setSearchType(QUERY_THEN_FETCH)
  11.         .setScroll(new TimeValue(scrollTimeout.toMillis()))
  13.         .setFetchSource(fields)
  15.         .setQuery(predict)
  17.         .setPreference("_shards:" + shard)
  19.         .setSize(size);            
  23. SearchRequest searchRequest searchRequestBuilder.request()
  29. SearchResponse searchResponse = null;
  31. try {
  33.     searchResponse = client.search(searchRequest);
  35. } catch (IOException e) {
  37.     LOG.error("Search has error: {}", e.getMessage());
  39. }
  41. if (searchResponse != null) {
  43.     currentScrollWindowId = searchResponse.getScrollId();
  47.     currentScrollWindowHits = searchResponse.getHits().getHits();
  51.     nextRecordIndex = 0;
  53. }

}

@Override

public ElasticsearchInputSplit[] createInputSplits(int minNumSplits) throws IOException {

  1.  return callBridge.createInputSplitsInternal(client, minNumSplits);

}

@Override

public InputSplitAssigner getInputSplitAssigner(ElasticsearchInputSplit[] inputSplits) {

  1.  return new LocatableInputSplitAssigner(inputSplits);

}

@Override

public boolean reachedEnd() throws IOException {

  1. if (limit > 0 && currentReadCount >= limit) {
  3.     return true;
  5. }
  9. if (currentScrollWindowHits.length > 0 && nextRecordIndex > currentScrollWindowHits.length - 1) {
  11.     fetchNextScrollWindow();
  13. }
  17. return currentScrollWindowHits.length == 0;

}

@Override

public Row nextRecord(Row reuse) throws IOException {

  1. if (reachedEnd()) {
  3.     LOG.warn("Already reached the end of the split.");
  5. }
  9. SearchHit hit = currentScrollWindowHits[nextRecordIndex];
  11. nextRecordIndex += 1;
  13. currentReadCount++;
  17. LOG.debug("Yielding new record for hit: {}", hit);
  21. return parseSearchHit(hit);

}

}

EsSource design doc - 图2

Adapt to different es versions

  • In order to adapt to different es versions(5 for TransportClient , 6,7 for RestHighLevelClient), we should add two methods in ElasticsearchApiCallBridge

interface ElasticsearchApiCallBridge {

//this is for getSearchShards for different es client

ElasticsearchInputSplit[] createInputSplitsInternal(String index, C client, int minNumSplits);

SearchResponse search(C client, SearchRequest searchRequest) throws IOException;

SearchResponse scroll(C client, SearchScrollRequest searchScrollRequest) throws IOException;

void close(C client) throws IOException;

}

  1. - take search as an exmaple
  2. -  for Elasticsearch5ApiCallBridge
  3. ```java
  4. //for Elasticsearch5ApiCallBridge
  5. @Override
  6. public SearchResponse search(TransportClient client, SearchRequest searchRequest) throws IOException {
  7.   //convert elasticsearchBuilder to SearchRequestBuiler
  8.   return client.search(searchRequest).actionGet();
  9. }
  • for Elasticsearch6ApiCallBridge and Elasticsearch7ApiCallBridge

//for Elasticsearch6ApiCallBridge and Elasticsearch7ApiCallBridge

@Override

public SearchResponse search(RestHighLevelClient client, SearchRequest searchRequest) throws IOException {

  1. //convert elasticsearchBuilder to SearchRequest

return client.search(searchRequest);

}

how to construct SearchRequest

//converts ElasticSearchBuilder to SearchRequestBuilder in es 5 for TransportClient api

SearchRequestBuilder searchRequestBuilder = client.prepareSearch(indices)

  1.     .setTypes(type)
  3.     .setSearchType(QUERY_THEN_FETCH)
  5.     .setScroll(new TimeValue(scrollTimeout.toMillis()))
  7.     .setFetchSource(fields.toArray(new String[0]), null)
  9.     .setQuery(BoolQueryBuilder)
  11.     .setPreference("_shards:" + shard)
  13.     .addSort("_doc", ASC)
  15.     .setSize(scrollSize);

SearchRequest searchRequest searchRequestBuilder.request()

DDL and configuration

  • for FLIP-122 New Connector Property Keys for New Factory
    • for es DynamicTableSource
  1. CREATE TABLE es_table (
  2.    ...
  3.   ) WITH (
  4.    'connector' = 'elasticsearch-7',
  5.    'hosts' = 'http://localhost:9092',
  6.    'index' = 'MyIndex',
  7.    'document-type' = 'MyType'
  8.    'scan.xx'= 'xx',
  9.   );

ElasticSearchTableSource

EsSource design doc - 图3

FilterPushDown and ProjectPushDown and LimitPushDown

FilterPushDown:
  • extract List, which can be pushdown to es and convert to es BoolQueryBuilder

@Override

public TableSource applyPredicate(List predicates) {

  1. BoolQueryBuilder builder = new BoolQueryBuilder();
  5. // try to convert Flink filter expressions to Parquet FilterPredicates
  7. List<Expression> unsupportedExpressions = new ArrayList<>(predicates.size());
  9. for (Expression toConvert : predicates) {
  11.     if(toElasticSearchQueryBuilder(toConvert, builder) == null) {
  13.         unsupportedExpressions.add(toConvert);
  15.     }
  17. }
  21. // update list of Flink expressions to unsupported expressions
  23. predicates.clear();
  25. predicates.addAll(unsupportedExpressions);
  29. // create and return a new ParquetTableSource with Parquet FilterPredicate
  31. return new ElasticsearchTableSource(schema, hosts, index, deserializer, sourceOptions, projectedFields, builder);

}

private boolean isValid(BinaryComparison comp) {

  1. return (comp.left() instanceof Literal && comp.right() instanceof Attribute) ||
  3.         (comp.left() instanceof Attribute && comp.right() instanceof Literal);

}

/**

  • Converts Flink Expression to ElasticSearch FilterPredicate.

    */

@Nullable

private QueryBuilder toElasticSearchQueryBuilder(Expression exp, BoolQueryBuilder builder) {

  1. if (exp instanceof Not) {
  3.     QueryBuilder c = toElasticSearchQueryBuilder(((Not) exp).child(), builder);
  5.     if (c == null) {
  7.         return null;
  9.     } else {
  11.         builder.mustNot(c);
  13.     }
  15. } else if (exp instanceof BinaryComparison) {
  17.     BinaryComparison binComp = (BinaryComparison) exp;
  21.     if (!isValid(binComp)) {
  23.         // unsupported literal Type
  25.         LOG.debug("Unsupported predict [{}] cannot be pushed to ElasticSearchTableSource.", exp);
  27.         return null;
  29.     }
  33.     boolean onRight = literalOnRight(binComp);
  37.     String columnName = getColumnName(binComp);
  39.     Object value = getLiteral(binComp);
  43.     if (exp instanceof EqualTo) {
  45.         return builder.must(new TermQueryBuilder(columnName, value));
  47.     } else if (exp instanceof NotEqualTo) {
  49.         return builder.mustNot(new TermQueryBuilder(columnName, value));
  51.     } else if (exp instanceof GreaterThan) {
  53.         if (onRight) {
  55.             return builder.must(greaterThan(columnName, value));
  57.         } else {
  59.             return builder.must(lessThan(columnName, value));
  61.         }
  63.     } else if (exp instanceof GreaterThanOrEqual) {
  65.         if (onRight) {
  67.             return builder.must(greaterThanOrEqual(columnName, value));
  69.         } else {
  71.             return builder.must(lessThanOrEqual(columnName, value));
  73.         }
  75.     } else if (exp instanceof LessThan) {
  77.         if (onRight) {
  79.             return builder.must(lessThan(columnName, value));
  81.         } else {
  83.             return builder.must(greaterThan(columnName, value));
  85.         }
  87.     } else if (exp instanceof LessThanOrEqual) {
  89.         if (onRight) {
  91.             return builder.must(lessThanOrEqual(columnName, value));
  93.         } else {
  95.             return builder.must(greaterThanOrEqual(columnName, value));
  97.         }
  99.     } else {
  101.         // Unsupported Predicate
  103.         LOG.debug("Unsupported predicate [{}] cannot be pushed into ElasticSearchTableSource.", exp);
  105.         return null;
  107.     }
  109. } else if (exp instanceof BinaryExpression) {
  111.     if (exp instanceof And) {
  113.         LOG.debug("All of the predicates should be in CNF. Found an AND expression: {}.", exp);
  115.     } else if (exp instanceof Or) {
  117.         QueryBuilder c1 = toElasticSearchQueryBuilder(((Or) exp).left(), builder);
  119.         QueryBuilder c2 = toElasticSearchQueryBuilder(((Or) exp).right(), builder);
  123.         if (c1 == null || c2 == null) {
  125.             return null;
  127.         } else {
  129.             return builder.should(c1).should(c2);
  131.         }
  133.     } else {
  135.         // Unsupported Predicate
  137.         LOG.debug("Unsupported predicate [{}] cannot be pushed into ParquetTableSource.", exp);
  139.         return null;
  141.     }
  143. } else if (exp instanceof UnaryExpression) {
  145.     String columnName = ((Attribute) ((UnaryExpression) exp).child()).name();
  149.     if (exp instanceof IsNull) {
  151.         builder.must(new ExistsQueryBuilder(columnName));
  153.     } else if (exp instanceof IsNotNull) {
  155.         builder.mustNot(new ExistsQueryBuilder(columnName));
  157.     }
  159. }
  161. return null;

}

private QueryBuilder greaterThan(String columnName, Object value) {

  1. return new RangeQueryBuilder(columnName).gt(value);

}

private QueryBuilder greaterThanOrEqual(String columnName, Object value) {

  1. return new RangeQueryBuilder(columnName).gte(value);

}

private QueryBuilder lessThan(String columnName, Object value) {

  1. return new RangeQueryBuilder(columnName).lt(value);

}

private QueryBuilder lessThanOrEqual(String columnName, Object value) {

  1. return new RangeQueryBuilder(columnName).lte(value);

}

private Object getLiteral(BinaryComparison comp) {

  1. if (literalOnRight(comp)) {
  3.     return ((Literal) comp.right()).value();
  5. } else {
  7.     return ((Literal) comp.left()).value();
  9. }

}

private String getColumnName(BinaryComparison comp) {

  1. if (literalOnRight(comp)) {
  3.     return ((Attribute) comp.left()).name();
  5. } else {
  7.     return ((Attribute) comp.right()).name();
  9. }

}

private boolean literalOnRight(BinaryComparison comp) {

  1. if (comp.left() instanceof Literal && comp.right() instanceof Attribute) {
  3.     return false;
  5. } else if (comp.left() instanceof Attribute && comp.right() instanceof Literal) {
  7.     return true;
  9. } else {
  11.     throw new RuntimeException("Invalid binary comparison.");
  13. }

}

@Override

public boolean isFilterPushedDown() {

  1. return queryBuilder != null;

}

LimitPushDown:

  • just transfer limit to ElasticSearchRowDataInputFormat

Test Plan

  • after discussion, we will implement it and add unit and ttegrate test
  • and we achieve the initial simple implementation for es7

Future plan

  • support es translog connector just like mysql binlog/hbase wal connector