java - 将数据从 cassandra 导出到 mongodb

Question

我有一个程序可以将数据直接从 Cassandra 导出到 mongodb。它工作正常但是，它不会将 counter type 列从 Cassandra 复制到 mongodb。它留空。

public class Export

{
static String keyspace = "db1";
static String table = "results";
static String host = "localhost";

@SuppressWarnings("deprecation")
static Mongo mongo = new Mongo("localhost", 27017);
@SuppressWarnings("deprecation")
static DB db = mongo.getDB("mydb");

static DBCollection collection = db.getCollection("results");

public static void main( String[] args ) throws IOException
{

    Cluster.Builder clusterBuilder = Cluster.builder()
     .addContactPoints(host);
     Cluster cluster = clusterBuilder.build();
     Session session = cluster.connect(keyspace);

     Statement stmt = new SimpleStatement("SELECT * FROM " + table);
     stmt.setFetchSize(1);
     ResultSet rs = session.execute(stmt);
     Iterator<Row> iter = rs.iterator();

     ArrayList<String> colName = new ArrayList<String>();
     Row row1 = iter.next();
     if (row1 != null)
     {
         for (Definition key1 : row1.getColumnDefinitions().asList())
         {      
                 String val = key1.getName();
                 colName.add(val);
         }

     }

     while (!rs.isFullyFetched()) 
     {
         rs.fetchMoreResults();
         Row row = iter.next();
         if (row != null)
         {   BasicDBObject document = new BasicDBObject();
             ArrayList<String> ele = new ArrayList<String>();
             for (Definition key : row.getColumnDefinitions().asList())
             {  
                 String val = myGetValue(key, row);
                 ele.add(val);
             }

             for (int i = 0; i < ele.size() && i < colName.size(); i++) {
                     document.put(colName.get(i), ele.get(i));
                }
                     collection.insert(document);
         }
     }

        session.close();
        cluster.close();

 }

    public static String myGetValue(Definition key, Row row)
     {
        String str = "";

        if (key != null)
        {
            String col = key.getName();

            try
            {
                if (key.getType() == DataType.cdouble())
                {
                    str = new Double(row.getDouble(col)).toString();
                }
                else if (key.getType() == DataType.cint())
                {
                    str = new Integer(row.getInt(col)).toString();
                }
                else if (key.getType() == DataType.uuid())
                {
                    str = row.getUUID(col).toString();
                }
                else if (key.getType() == DataType.cfloat())
                {
                    str = new Float(row.getFloat(col)).toString();
                }
                else if (key.getType() == DataType.counter())
                {
                    str = new Float(row.getFloat(col)).toString();
                }

                else if (key.getType() == DataType.timestamp())
                {
                    str = row.getDate(col).toString();

                    SimpleDateFormat fmt = new SimpleDateFormat("yyyy-MM-dd HH:mm:ssZ");
                    str = fmt.format(row.getDate(col));
                }
                else
                {
                    str = row.getString(col);
                }
            } catch (Exception e)
                {
                    str = "";
                }
         }

                return str;
       }

    }

导出到 mongodb 的数据是:

{
"_id" : ObjectId("558c0202209c02284d30df05"),
"term" : "gamma red eye tennis dampener",
"year" : "2015",
"month" : "05",
"day" : "29",
"hour" : "09",
"dayofyear" : "176",
"weekofyear" : "26",
"productcount" : "1",
"count" : ""
}
{
"_id" : ObjectId("558c0202209c02284d30df06"),
"term" : "headtie",
"year" : "2015",
"month" : "06",
"day" : "01",
"hour" : "05",
"dayofyear" : "176",
"weekofyear" : "26",
"productcount" : "29",
"count" : ""
}
{
"_id" : ObjectId("558c0202209c02284d30df07"),
"term" : "dryed roller court",
"year" : "2015",
"month" : "06",
"day" : "08",
"hour" : "15",
"dayofyear" : "176",
"weekofyear" : "26",
"productcount" : "362",
"count" : ""
 }

我知道问题出在哪里。下面的函数没有将计数器的值插入到 mongodb 中，因为 val 被定义为与 Cassandra 的计数器数据类型冲突的字符串。

while (!rs.isFullyFetched()) 
     {
         rs.fetchMoreResults();
         Row row = iter.next();
         if (row != null)
         {   BasicDBObject document = new BasicDBObject();
             ArrayList<String> ele = new ArrayList<String>();
             for (Definition key : row.getColumnDefinitions().asList())
             {  
                 String val = myGetValue(key, row);
                 System.out.println(val);
                 ele.add(val);
             }

             for (int i = 0; i < ele.size() && i < colName.size(); i++) {
                     document.put(colName.get(i), ele.get(i));
                }
                     collection.insert(document);
         }
     }

任何人都可以建议我需要在函数中做的必要更改吗？

Answer 1

您应该使用 getLong() 而不是 getFloat() 来检索计数器值，即更改代码:

else if (key.getType() == DataType.counter())
{
     str = new Float(row.getFloat(col)).toString();
}

到

else if (key.getType() == DataType.counter())
{
     str = String.valueOf(row.getLong(col));
}

顺便说一句，请注意您不必创建包装器类型 Long 的实例(就像您在所有案例中所做的那样)。使用 String.valueOf() 创建值的字符串表示形式。