Hbase Upsert with Spark

Question

나는 spark 스트리밍 작업을하고 있으며 일부는 일부 집계 작업을 수행 중이다. 이제는 HBase에 레코드를 삽입하고 싶다. 전형적인 삽입이 아니다. rowkey에 대한 컬럼 값이 합계보다 많으면 UPSERT를하고 싶다. (newvalue + oldvalue)가 발생해야합니다. 아무도 자바에서 의사 코드를 공유합니까? 어떻게해야합니까?

Answer 1

이런 식으로 뭔가 ...

byte[] rowKey = null; // Provided 
Table table = null; // Provided 
long newValue = 1000; // Provided 
byte[] FAMILY = new byte[]{0}; // Defined 
byte[] QUALIFIER = new byte[]{1}; // Defined 

try { 
    Get get = new Get(rowKey); 
    Result result = table.get(get); 
    if (!result.isEmpty()) { 
     Cell cell = result.getColumnLatestCell(FAMILY, QUALIFIER); 
     newValue += Bytes.bytesToLong(cell.getValueArray(),cell.getValueOffset()); 
    } 
    Put put = new Put(rowKey); 
    put.addColumn(FAMILY,QUALIFIER,Bytes.toBytes(newValue)); 
    table.put(put); 
} catch (Exception e) { 
    // Handle Exceptions... 
} 

우리는 (스플 라이스 기계 [오픈 소스]) HBase를에 데이터를 저장하는 스파크 스트리밍을 사용하여 꽤 멋진 튜토리얼이있다.

체크 아웃 it. 흥미로울 수 있습니다.

Answer 2

나는 아래의 방법은 의사 코드입니다 발견 : - UPSERT를 들어

=========== (업데이트 및 삽입) ===========

공공 무효 HbaseUpsert (JavaRDD < 행> javaRDD는) ========== IOException이, ServiceException {구성

 JavaPairRDD < ImmutableBytesWritable, Put > hbasePuts1 = javaRDD.mapToPair(

     new PairFunction < Row, ImmutableBytesWritable, Put >() { 

     private static final long serialVersionUID = 1L; 
    public Tuple2 < ImmutableBytesWritable, Put > call(Row row) throws Exception { 

      if(HbaseConfigurationReader.getInstance()!=null) 
      { 
      HTable table = new HTable(HbaseConfigurationReader.getInstance().initializeHbaseConfiguration(), "TEST"); 

     try { 

      String Column1 = row.getString(1); 
      long Column2 = row.getLong(2); 
      Get get = new Get(Bytes.toBytes(row.getString(0))); 
       Result result = table.get(get); 
       if (!result.isEmpty()) { 
        Cell cell = result.getColumnLatestCell(Bytes.toBytes("cf1"), Bytes.toBytes("Column2")); 
        Column2 += Bytes.toLong(cell.getValueArray(),cell.getValueOffset()); 
       } 
      Put put = new Put(Bytes.toBytes(row.getString(0))); 
      put.add(Bytes.toBytes("cf1"), Bytes.toBytes("Column1"), Bytes.toBytes(Column1)); 
      put.add(Bytes.toBytes("cf1"), Bytes.toBytes("Column2"), Bytes.toBytes(Column2)); 
      return new Tuple2 < ImmutableBytesWritable, Put > (new ImmutableBytesWritable(), put); 

     } catch (Exception e) { 

      e.printStackTrace(); 
     } 
     finally { 
      table.close(); 
     } 
      } 
     return null; 
     } 
     }); 

    hbasePuts1.saveAsNewAPIHadoopDataset(HbaseConfigurationReader.initializeHbaseConfiguration()); 

    } 

==============를 던졌습니다 ===== 공용 클래스 HbaseConfigurationReader 구현 가능 Serializable {

static Job newAPIJobConfiguration1 =null; 
private static Configuration conf =null; 
private static HTable table= null; 
private static HbaseConfigurationReader instance= null; 

private static Log logger= LogFactory.getLog(HbaseConfigurationReader.class); 

HbaseConfigurationReader()는 MasterNotRunningException, ZooKeeperConnectionException, ServiceException, IOException { initializeHbaseConfiguration()을 발생시킵니다. }

공용 static HbaseConfigurationReader의 getInstance()가 MasterNotRunningException, ZooKeeperConnectionException, ServiceException IOException을 {

if (instance == null) { 
    instance = new HbaseConfigurationReader(); 
} 

return instance; 

} 공개 정적 구성 initializeHbaseConfiguration() MasterNotRunningException, ZooKeeperConnectionException, ServiceException을 발생 IOException을 { 을 던지면 (conf의 = = null) { conf = HBaseConfiguration.create(); conf.set ("hbase.zookeeper.quorum", "localhost"); conf.set ("hbase.zookeeper.property.clientPort", "2181"); HBaseAdmin.checkHBaseAvailable (conf); table = new HTable (conf, "TEST"); conf.set (org.apache.hadoop.hbase.mapreduce.TableInputFormat.INPUT_TABLE, "TEST"); 시도 { newAPIJobConfiguration1 = Job.getInstance (conf); newAPIJobConfiguration1.getConfiguration(). set (TableOutputFormat.OUTPUT_TABLE, "TEST"); newAPIJobConfiguration1.setOutputFormatClass (org.apache.hadoop.hbase.mapreduce.TableOutputFormat.class); } catch (IOException e) { e.printStackTrace(); }

} 

else 
    logger.info("Configuration comes null"); 

return newAPIJobConfiguration1.getConfiguration(); 

} } 또한