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。