bin : record.getBins().entrySet()) {
if (bin.getValue() instanceof String) {
payloadBuilder.append(bin.getKey());
payloadBuilder.append(separator);
payloadBuilder.append(bin.getValue());
payloadBuilder.append(System.lineSeparator());
}
}
return new DefaultTextOutboundRecord<>(
payloadBuilder.toString().getBytes(), MediaType.OCTET_STREAM,
formattedRecord.getMetadata());
}
}
```
An example batch format transformer for Elasticsearch outbound destination.
```yaml
...
namespaceA:
format:
mode: custom
batch-formatter-class: com.aerospike.connect.outbound.transformer.examples.elasticsearch.ElasticsearchCustomJsonBatchFormatter
```
```java
package com.aerospike.connect.outbound.transformer.examples.elasticsearch;
import co.elastic.clients.elasticsearch.core.BulkRequest.Builder;
import co.elastic.clients.elasticsearch.core.bulk.CreateOperation;
import com.aerospike.connect.outbound.elasticsearch.ElasticsearchOutboundMetadata;
import com.aerospike.connect.outbound.elasticsearch.format.ElasticsearchOutboundRecord;
import com.aerospike.connect.outbound.format.BatchFormatter;
import com.aerospike.connect.outbound.format.BatchItem;
import com.aerospike.connect.outbound.format.OutboundRecord;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.NonNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import javax.inject.Singleton;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
/**
* ElasticsearchCustomJsonBatchFormatter formats change notification records by
* adding an extra field named 'created_at' along with the parsed bins. Also,
* setting the index at operation level if it can be derived.
*
*
* A snippet of a config for this batch-formatter can be
*
* format:
* mode: custom
* batch-formatter-class: com.aerospike.connect.outbound.transformer.examples.elasticsearch.ElasticsearchCustomJsonBatchFormatter
*
*/
@Singleton
public class ElasticsearchCustomJsonBatchFormatter
implements BatchFormatter {
private final static Logger logger =
LoggerFactory.getLogger(
ElasticsearchCustomJsonBatchFormatter.class.getName());
private final static ObjectMapper objectMapper = new ObjectMapper();
@Override
public List> format(
@NonNull List> batchItems)
throws Exception {
ElasticsearchOutboundRecord elasticsearchOutboundRecord =
getElasticsearchOutboundRecord(batchItems);
List>
outboundRecords = new ArrayList<>();
outboundRecords.add(elasticsearchOutboundRecord);
return outboundRecords;
}
@NonNull
private ElasticsearchOutboundRecord getElasticsearchOutboundRecord(
@NonNull List> batchItems) {
long currentTimeMillis = System.currentTimeMillis();
return () -> {
Builder builder = new Builder();
// This is the default index for all bulk commands if we don't
// override at the operation level.
builder.index("my_test_index");
batchItems.stream().map(BatchItem::getRecord).forEach(record -> {
logger.debug("Formatting record {}",
record.getMetadata().getKey());
try {
Map bins = record.getBins();
Map resultMap =
new HashMap<>(bins.size() + 1);
resultMap.putAll(bins);
resultMap.put("created_at", currentTimeMillis);
byte[] value =
objectMapper.writeValueAsBytes(resultMap);
builder.operations(
op -> op.create(CreateOperation.of(cob -> {
cob.document(
new PreSerializedJson(value));
Object esIndex = bins.get("es_index");
if (esIndex instanceof String) {
// Set index for this Create
// operation.
cob.index((String) esIndex);
}
return cob;
})
));
} catch (Exception e) {
throw new RuntimeException(e);
}
}
);
return builder.build();
};
}
}
```
## Examples
For more examples see [examples](https://github.com/aerospike/aerospike-connect-outbound-sdk/tree/aerospike-connect-outbound-sdk-3.0.0/examples) in the SDK GitHub repo.