Hbase Table.batch занимает 300 секунд, чтобы вставить 800 000 записей в таблицу

Question

Я читаю json-файл размером 30 мб, обрабатываю для создания семейства столбцов и значений ключей. Затем создайте объект Put, вставьте rowkey и значения в него. Создайте список таких объектов put и вызовите Table.batch() и передайте этот список. Я называю это, когда мой размер arraylist составляет 50000. Затем очистите список и вызовите следующую партию. Однако обрабатывать файл, который в итоге имеет 800 000 записей, занимает 300 секунд. Я тоже устал table.put, но он был еще медленнее. Я использую hbase 1.1. Я получаю этот json от Kafka. Приветствуются любые предложения по повышению производительности. Я проверил SO-форумы, но не очень помог. Я поделюсь кодом, если вы хотите взглянуть на него.

С уважением

Raghavendra

public static void processData(String jsonData) 
{ 
    if (jsonData == null || jsonData.isEmpty()) 
    { 
     System.out.println("JSON data is null or empty. Nothing to process"); 
     return; 
    } 

    long startTime = System.currentTimeMillis(); 

    Table table = null; 
    try 
    { 
     table = HBaseConfigUtil.getInstance().getConnection().getTable(TableName.valueOf("MYTABLE")); 
    } 
    catch (IOException e1) 
    { 
     System.out.println(e1); 
    } 

    Put processData = null; 
    List<Put> bulkData = new ArrayList<Put>(); 

    try 
    { 

     //Read the json and generate the model into a class  
     //ProcessExecutions is List<ProcessExecution> 
     ProcessExecutions peData = JsonToColumnData.gson.fromJson(jsonData, ProcessExecutions.class); 

     if (peData != null) 
     { 
      //Read the data and pass it to Hbase 
      for (ProcessExecution pe : peData.processExecutions) 
      { 
       //Class Header stores some header information 
       Header headerData = pe.getHeader(); 

       String rowKey = headerData.getRowKey(); 
       processData = new Put(Bytes.toBytes(JsonToColumnData.rowKey)); 
       processData.addColumn(Bytes.toBytes("Data"), 
           Bytes.toBytes("Time"), 
           Bytes.toBytes("value")); 

       //Add to list 
       bulkData.add(processData);    
       if (bulkData.size() >= 50000) //hardcoded for demo 
       { 
        long tmpTime = System.currentTimeMillis(); 
        Object[] results = null; 
        table.batch(bulkData, results);      
        bulkData.clear(); 
        System.gc(); 
       } 
      } //end for 
      //Complete the remaining write operation 
      if (bulkData.size() > 0) 
      { 
       Object[] results = null; 
       table.batch(bulkData, results); 
       bulkData.clear(); 
       //Try to free memory 
       System.gc(); 
      } 
    } 
    catch (Exception e) 
    { 
     System.out.println(e); 
     e.printStackTrace(); 
    } 
    finally 
    { 
     try 
     { 
      table.close(); 
     } 
     catch (IOException e) 
     { 
      System.out.println("Error closing table " + e); 
      e.printStackTrace(); 
     } 
    } 

} 


//This function is added here to show the connection 
/*public Connection getConnection() 
{ 

    try 
    { 
     if (this.connection == null) 
     { 
      ExecutorService executor = Executors.newFixedThreadPool(HBaseConfigUtil.THREADCOUNT); 
      this.connection = ConnectionFactory.createConnection(this.getHBaseConfiguration(), executor); 
     } 
    } 
    catch (IOException e) 
    { 
     e.printStackTrace(); 
     System.out.println("Error in getting connection " + e.getMessage()); 
    } 

    return this.connection; 
}*/ 

Answer 1

У меня был тот же случай, когда мне нужно разобрать 5 ГБ JSON и вставить в Hbase таблицу ... Вы можете попробовать ниже способ (который должен работать), который оказался очень быстрым для партии 100000 записей в моем случае.

public void addMultipleRecordsAtaShot(final ArrayList<Put> puts, final String tableName) throws Exception { 
     try { 
      final HTable table = new HTable(HBaseConnection.getHBaseConfiguration(), getTable(tableName)); 
      table.put(puts); 
      LOG.info("INSERT record[s] " + puts.size() + " to table " + tableName + " OK."); 
     } catch (final Throwable e) { 
      e.printStackTrace(); 
     } finally { 
      LOG.info("Processed ---> " + puts.size()); 
      if (puts != null) { 
       puts.clear(); 
      } 
     } 
    } 

Для получения дополнительной информации, чтобы увеличить размер буфера проверить мой ответ в другом context увеличить размер буфера см Doc https://hbase.apache.org/apidocs/org/apache/hadoop/hbase/client/Table.html