Карта таблицы базы данных cassandra с использованием искры и RDD

Question

Мне нужно сопоставить таблицу, в которой написана история использования приложения. Таблица получила эти кортежи:

<AppId,date,cpuUsage,memoryUsage> 
<AppId,date,cpuUsage,memoryUsage> 
<AppId,date,cpuUsage,memoryUsage> 
<AppId,date,cpuUsage,memoryUsage> 
<AppId,date,cpuUsage,memoryUsage> 

AppId всегда отличается, потому что упоминается у многих приложений, date выражается в этом формате dd/mm/yyyy hh/mmcpuUsage и memoryUsage выражаются в % так, например:

<3ghffh3t482age20304,230720142245,0.2,3,5> 

Я извлек данные из кассандры таким образом (небольшой фрагмент):

public static void main(String[] args) { 
     Cluster cluster; 
     Session session; 
     cluster = Cluster.builder().addContactPoint("127.0.0.1").build(); 
     session = cluster.connect(); 
     session.execute("CREATE KEYSPACE IF NOT EXISTS foo WITH replication " 
       + "= {'class':'SimpleStrategy', 'replication_factor':3};"); 
     String createTableAppUsage = "CREATE TABLE IF NOT EXISTS foo.appusage" 
       + "(appid text,date text, cpuusage double, memoryusage double, " 
       + "PRIMARY KEY(appid,date) " + "WITH CLUSTERING ORDER BY (time ASC);"; 
     session.execute(createTableAppUsage); 
     // Use select to get the appusage's table rows 
     ResultSet resultForAppUsage = session.execute("SELECT appid,cpuusage FROM foo.appusage"); 
     for (Row row: resultForAppUsage) 
      System.out.println("appid :" + row.getString("appid") +" "+ "cpuusage"+row.getString("cpuusage")); 
     // Clean up the connection by closing it 
     cluster.close(); 
    } 

Итак, моя задача теперь сопоставить данные по key value и создать кортеж интегрирующую этот код (фрагмент кода, который не работает):

 <AppId,cpuusage> 

     JavaPairRDD<String, Integer> saveTupleKeyValue =someStructureFromTakeData.mapToPair(new PairFunction<String, String, Integer>() { 
      public Tuple2<String, Integer> call(String x) { 
       return new Tuple2(x, y); 
      } 

как я могу отобразить APPID и cpuusage используя RDD и уменьшить eg. cpuusage >50?

Любая помощь?

благодарит заранее.

Answer 1

Если предположить, что у вас есть действительная SparkContext sparkContext уже создали, добавила зависимости коннекторов искровой Кассандры к вашему проекту и настроит искровое приложение, чтобы поговорить с вашей Кассандрой кластера (см docs для этого), то мы можем загрузить данные в РДУ, как это:

val data = sparkContext.cassandraTable("foo", "appusage").select("appid", "cpuusage") 

в Java, идея та же, но она требует немного больше сантехники, описанного here