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

Z时代
2024-01-10
分类：综合

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");
	}
}