将MySQL一张表的数据迁移到MongoDB数据库的Java代码示例

Java代码：

package com.zifeiy.snowflake.handle.etl.mongodb;

import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.List;

import org.bson.Document;

import com.mongodb.MongoClient;
import com.mongodb.client.FindIterable;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoCursor;
import com.mongodb.client.MongoDatabase;
import com.zifeiy.snowflake.handle.etl.dbconnection.DatabaseConnectionFactory;

/*
 * 这个类用于将MySQL中的一张表的数据原模原样地传输到MongoDB中
 */
public class MySQLtoMongoDBDataTransfer {
    
    public boolean transfer(String databaseName, String tableName) {
        try {
            // mysql connection
            Connection connection = DatabaseConnectionFactory.createConnection(databaseName);
            Statement statement = connection.createStatement();
            ResultSet resultSet = statement.executeQuery("select * from " + tableName);
            // mongo connect
            MongoClient mongoClient = new MongoClient();
            MongoDatabase mongoDatabase = mongoClient.getDatabase(databaseName);
            MongoCollection<Document> collection = mongoDatabase.getCollection(tableName);
            // 首先清空collection
            FindIterable<Document> deleteDocuments = collection.find();
            MongoCursor<Document> deleteIterator = deleteDocuments.iterator();
            while (deleteIterator.hasNext()) {
                collection.deleteOne(deleteIterator.next());
            }
            // begin transfer
            int columnCount = -1, count = 0;
            List<Document> documents = new ArrayList<Document>();
            while (resultSet.next()) {
                if (columnCount == -1) {
                    columnCount = resultSet.getMetaData().getColumnCount();
                }
                Document document = new Document();
                for (int i = 1; i <= columnCount; i ++) {
                    String columnName = resultSet.getMetaData().getColumnName(i);
                    Object value = resultSet.getObject(i);
                    document.append(columnName, value);
                }
                documents.add(document);
                count ++;
                if (count % 1000 == 0) {    // 每1000条数据执行一次插入
                    collection.insertMany(documents);
                    documents = new ArrayList<Document>();
                    System.out.println("transfer " + count + " lines from mysql to mongodb  ...");
                }
            }
            System.out.println("total transfer " + count + " lines!");
            
            DatabaseConnectionFactory.closeConnection(connection);
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }
    
    // main for test
    public static void main(String[] args) {
        new MySQLtoMongoDBDataTransfer().transfer("naturaldb", "NAT_IMBS_T_SIGN");
    }
}

其中：

• databaseName对应MySQL数据库的database，对应MongoDB的database；
  ` tableName对应MySQL数据库的表，对应MongoDB的collection。