Java代码：将PostgreSQL表数据导入到Elasticsearch索引

￡神魔★判官ぃ 2023-10-14 22:04 12阅读 0赞

对于Elasticsearch而言，它并不会直接按照PostgreSQL中的表一一创建索引。在将数据从PostgreSQL导入到Elasticsearch时，您可以根据需求和数据模型设计，决定如何在Elasticsearch中进行索引的创建和映射。

一种常见的做法是为每个表创建一个对应的索引，并将表中的每行数据映射为Elasticsearch中的文档。您可以根据表的结构和字段定义，在Elasticsearch中创建相应的索引和映射，以便正确存储和搜索数据。

以下是一个示例代码，展示如何为一个名为`your_table`的PostgreSQL表创建对应的Elasticsearch索引，并将数据导入到该索引中：

## 依赖 ##

对于PostgreSQL 12.4，您可以使用以下pom.xml配置添加PostgreSQL JDBC驱动程序的依赖项：

<dependencies>
        
        <dependency>
            <groupId>org.postgresql</groupId>
            <artifactId>postgresql</artifactId>
            <version>42.2.23</version>
        </dependency>
    
        
        <dependency>
            <groupId>org.elasticsearch.client</groupId>
            <artifactId>elasticsearch-rest-high-level-client</artifactId>
            <version>7.15.0</version>
        </dependency>
    </dependencies>

请确保将上述依赖项添加到您项目的pom.xml文件中，并根据需要进行其他配置和调整。这将确保您的项目在编译和运行时能够正确引入和使用PostgreSQL JDBC驱动程序和Elasticsearch高级客户端。

## 代码 ##

import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
    import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
    import org.elasticsearch.action.bulk.BulkRequest;
    import org.elasticsearch.action.bulk.BulkResponse;
    import org.elasticsearch.action.index.IndexRequest;
    import org.elasticsearch.client.RequestOptions;
    import org.elasticsearch.client.RestClient;
    import org.elasticsearch.client.RestHighLevelClient;
    import org.postgresql.core.BaseConnection;
    
    import java.sql.*;
    import java.io.IOException;
    
    public class PgToEsImporter {
        
        private static final String PG_HOST = "localhost";
        private static final int PG_PORT = 5432;
        private static final String PG_DATABASE = "your_pg_database";
        private static final String PG_USERNAME = "your_pg_username";
        private static final String PG_PASSWORD = "your_pg_password";
    
        private static final String ES_HOST = "localhost";
        private static final int ES_PORT = 9200;
        private static final String ES_INDEX = "your_es_index";
        private static final String ES_TYPE = "your_es_type";
    
        public static void main(String[] args) {
        
            try {
        
                // 连接到PostgreSQL数据库
                String pgUrl = "jdbc:postgresql://" + PG_HOST + ":" + PG_PORT + "/" + PG_DATABASE;
                Connection connection = DriverManager.getConnection(pgUrl, PG_USERNAME, PG_PASSWORD);
    
                // 连接到Elasticsearch
                RestHighLevelClient esClient = new RestHighLevelClient(
                        RestClient.builder(new HttpHost(ES_HOST, ES_PORT, "http")));
    
                // 创建Elasticsearch索引
                CreateIndexRequest createIndexRequest = new CreateIndexRequest(ES_INDEX);
                createIndexRequest.mapping(ES_TYPE, "{ \"properties\": { \"name\": { \"type\": \"text\" } } }");
                CreateIndexResponse createIndexResponse = esClient.indices().create(createIndexRequest, RequestOptions.DEFAULT);
                if (!createIndexResponse.isAcknowledged()) {
        
                    throw new IOException("Failed to create Elasticsearch index");
                }
    
                // 查询PostgreSQL数据库中的数据
                Statement statement = connection.createStatement();
                String query = "SELECT * FROM your_table";
                ResultSet resultSet = statement.executeQuery(query);
    
                // 创建批量请求
                BulkRequest bulkRequest = new BulkRequest();
    
                // 遍历结果集并将数据添加到批量请求中
                while (resultSet.next()) {
        
                    String id = resultSet.getString("id");
                    String name = resultSet.getString("name");
    
                    // 创建Elasticsearch索引请求
                    IndexRequest indexRequest = new IndexRequest(ES_INDEX, ES_TYPE, id)
                            .source("name", name);
    
                    // 添加索引请求到批量请求
                    bulkRequest.add(indexRequest);
                }
    
                // 执行批量请求
                BulkResponse bulkResponse = esClient.bulk(bulkRequest, RequestOptions.DEFAULT);
    
                // 处理批量请求的响应结果
                if (bulkResponse.hasFailures()) {
        
                    System.err.println("批量请求执行失败：" + bulkResponse.buildFailureMessage());
                } else {
        
                    System.out.println("数据导入成功！");
                }
    
                // 关闭资源
                resultSet.close();
                statement.close();
                connection.close();
                esClient.close();
            } catch (SQLException | IOException e) {
        
                e.printStackTrace();
            }
        }
    }

在上述示例中，我们在Elasticsearch中创建了一个名为`your_es_index`的索引，指定了一个名为`your_es_type`的文档类型，并为`name`字段定义了一个文本类型的映射。您可以根据自己的表结构和字段定义，调整和扩展Elasticsearch索引的创建和映射部分。

请确保将示例代码中的以下部分替换为您的实际信息：

*  `PG_DATABASE`：您的PostgreSQL数据库名称。
 *  `PG_USERNAME`：您的PostgreSQL数据库用户名。
 *  `PG_PASSWORD`：您的PostgreSQL数据库密码。
 *  `ES_INDEX`：您希望在Elasticsearch中创建的索引名称。
 *  `ES_TYPE`：您希望在Elasticsearch中创建的文档类型名称。

此示例代码使用了PostgreSQL JDBC驱动程序和Elasticsearch的高级客户端。请确保根据先前提供的POM配置中的依赖项，正确设置您的项目以便编译和运行代码。此代码将从PostgreSQL表中查询数据，并将其映射到Elasticsearch索引中的文档中。