使用Elasticsearch索引现有的关系数据库，例如Oracle

Question

我创建了与Elasticsearch应用集成的Spring Boot。现在可以为我自己的记录创建索引，该记录正在postman中输入，并且能够在id的帮助下搜索这些记录。

如何为我的Oracle DB的记录创建索引，我想在ID的帮助或这些记录的任何其他列值的帮助下搜索这些记录。

请在下面找到我现有的代码。

book.java

@JsonInclude(JsonInclude.Include.NON_EMPTY)
public class Book {
    private String id;
    private String title;
    public String getId() {
        return id;
    }
    public void setId(String id) {
        this.id = id;
    }
    public String getTitle() {
        return title;
    }
    public void setTitle(String title) {
        this.title = title;
    }
}

elasticsearchConfiguration.java

@Configuration
public class ElasticSearchConfiguration extends AbstractFactoryBean<RestHighLevelClient> {
    private static final Logger logger = LoggerFactory.getLogger(ElasticSearchConfiguration.class);
    @Value("${spring.data.elasticsearch.cluster-nodes}")
    private String clusterNodes;
    @Value("${spring.data.elasticsearch.cluster-name}")
    private String clusterName;
    private RestHighLevelClient restHighLevelClient;
    @Override
    public void destroy() {
        try {
            if (restHighLevelClient != null) {
                restHighLevelClient.close();
            }
        } catch (final Exception e) {
            logger.error("Error closing ElasticSearch client: ", e);
        }
    }
    @Override
    public Class<RestHighLevelClient> getObjectType() {
        return RestHighLevelClient.class;
    }
    @Override
    public boolean isSingleton() {
        return false;
    }
    @Override
    public RestHighLevelClient createInstance() {
        return buildClient();
    }
    private RestHighLevelClient buildClient() {
        try {
            restHighLevelClient = new RestHighLevelClient(
                    RestClient.builder(
                            new HttpHost("localhost", 9200, "http"),
                            new HttpHost("localhost", 9201, "http")));
        } catch (Exception e) {
            logger.error(e.getMessage());
        }
        return restHighLevelClient;
    }

}

bookcontroller.java

@RestController
@RequestMapping("/books")
public class BookController {
    private BookDao bookDao;
    public BookController(BookDao bookDao) {
        this.bookDao = bookDao;
    }
    @PostMapping
    public Book insertBook(@RequestBody Book book) throws Exception{
        return bookDao.insertBook(book);
    }
    @GetMapping("/{id}")
    public Map<String, Object> getBookById(@PathVariable String id){
        return bookDao.getBookById(id);
    }
    }
}

bookdao.java

@Repository
public class BookDao {
    private final String INDEX = "bookdata";
    private final String TYPE = "books";
    private RestHighLevelClient restHighLevelClient;
    private ObjectMapper objectMapper;
    public BookDao( ObjectMapper objectMapper, RestHighLevelClient restHighLevelClient) {
        this.objectMapper = objectMapper;
        this.restHighLevelClient = restHighLevelClient;
    }
    public Book insertBook(Book book){
        book.setId(UUID.randomUUID().toString());
        Map<String, Object> dataMap = objectMapper.convertValue(book, Map.class);
        IndexRequest indexRequest = new IndexRequest(INDEX, TYPE, book.getId())
                .source(dataMap);
        try {
            IndexResponse response = restHighLevelClient.index(indexRequest);
        } catch(ElasticsearchException e) {
            e.getDetailedMessage();
        } catch (java.io.IOException ex){
            ex.getLocalizedMessage();
        }
        return book;
    }
    public Map<String, Object> getBookById(String id){
        GetRequest getRequest = new GetRequest(INDEX, TYPE, id);
        GetResponse getResponse = null;
        try {
            getResponse = restHighLevelClient.get(getRequest);
        } catch (java.io.IOException e){
            e.getLocalizedMessage();
        }
        Map<String, Object> sourceAsMap = getResponse.getSourceAsMap();
        return sourceAsMap;
    }
    }
}

maven depencencies

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
    <!-- Elasticsearch Dependencies -->
    <dependency>
        <groupId>org.elasticsearch</groupId>
        <artifactId>elasticsearch</artifactId>
        <version>6.1.2</version>
    </dependency>
    <dependency>
        <groupId>org.elasticsearch.client</groupId>
        <artifactId>elasticsearch-rest-high-level-client</artifactId>
        <version>6.1.2</version>
    </dependency>
    <dependency>
        <groupId>org.apache.logging.log4j</groupId>
        <artifactId>log4j-api</artifactId>
        <version>2.8.2</version>
    </dependency>
    <dependency>
        <groupId>org.apache.logging.log4j</groupId>
        <artifactId>log4j-core</artifactId>
        <version>2.8.2</version>
    </dependency>
    <dependency>
        <groupId>org.elasticsearch.client</groupId>
        <artifactId>elasticsearch-rest-client</artifactId>
        <version>6.1.2</version>
    </dependency>
    <dependency>
        <groupId>org.elasticsearch.client</groupId>
        <artifactId>elasticsearch-rest-client-sniffer</artifactId>
        <version>6.1.2</version>
    </dependency>
</dependencies>

请找到我的输入和输出：

am以下格式通过Postman插入记录：

输入：

 {
    "title": "Java Always",
}

输出：

{
    "id": "be2e497f-585f-4dff-8d07-fbcee1dc1c80",
    "title": "Java Always",
}

Answer 1

我建议使用logstash。这是用于数据收集和管道的弹性产品。它具有JDBC输入，然后可以路由到Elasticsearch输出。

示例文件可能看起来像：

input {
  jdbc {
    jdbc_driver_library => "ojdbc6.jar"
    jdbc_driver_class => "oracle.jdbc.OracleDriver"
    jdbc_connection_string => "jdbc:oracle:thin:@localhost:1521/mydb"
    jdbc_user => "user"
    jdbc_password => "password"
    statement => "SELECT * FROM my_table"
  }
}
output {
  elasticsearch {
    hosts => ["localhost:9200"]
    index => "my_index"
  }
}

这应该在my_table中采用您所拥有的任何东西，并将其放入称为my_index的索引中。您有很多选项(例如，在时间表上运行DB查询(，因此请查看链接。