[HADOOP] 읽고 FLINK 스트리밍 작업에서 HBase를로 작성하는 방법

읽고 FLINK 스트리밍 작업에서 HBase를로 작성하는 방법

우리가 읽고 스트리밍 응용 프로그램에서 HBase를 쓸 경우 우리는 그것을 할 수있는 방법에 대해 설명합니다. 우리는 우리가 읽기에 대한 연결을 열 수있는 방법을 쓰기 위해 열려 방법을 통해 연결을 엽니 다.

object test {

    if (args.length != 11) {
      //print args
      System.exit(1)
    }

    val Array() = args
    println("Parameters Passed " + ...);

    val env = StreamExecutionEnvironment.getExecutionEnvironment


    val properties = new Properties()
    properties.setProperty("bootstrap.servers", metadataBrokerList)
    properties.setProperty("zookeeper.connect", zkQuorum)
    properties.setProperty("group.id", group)


    val messageStream = env.addSource(new FlinkKafkaConsumer08[String](topics, new SimpleStringSchema(), properties))

    messageStream.map { x => getheader(x) }





    def getheader(a: String) {

        //Get header and parse and split the headers
                if (metadata not available hit HBASE) { //Device Level send(Just JSON)

            //How to read from HBASE here .

                      } 
                      //If the resultset is not available in Map fetch from phoenix
                      else {
                          //fetch from cache
                      }
     }




    }
   messageStream.writeUsingOutputFormat(new HBaseOutputFormat());
   env.execute()

}

이제 방법 getheader 내부 내가 그렇게 할 수있는 방법을 경우 (메타 데이터를 사용할 수 없습니다 히트 HBase를) 내부 HBase를 읽고 싶은 경우. 내가 여기에 연결을 열 싶지 않아, 아이디어는 스레드에 대한 단일 연결을 유지하고 FLINK는 HBase와 함께하는 것처럼 open () 메서드 또는 얼마나 불꽃 foreachpartition와 함께 않습니다 싱크대, 것을 재사용하는 것입니다. 나는이 시도하지만 난 방법에 StreamExecutionEnvironment를 전달할 수 없습니다. 어떻게하면 사람이 조각을 제공 할 수,이를 수 있을까?

해결법

1. ==============================

   1.당신은 / 읽기 스트리밍 사용자 함수에서 아파치 HBase에 기록합니다. 당신이 연결된 HBaseReadExample은 다른 일을하고있다 : 그것은 데이터 집합 (FLINK의 일괄 처리 추상화) 내로 HBase를 테이블을 읽습니다. 사용자 기능이 코드를 사용하면 FLINK 프로그램 내에서 FLINK 프로그램을 시작하는 것을 의미한다.

   당신은 / 읽기 스트리밍 사용자 함수에서 아파치 HBase에 기록합니다. 당신이 연결된 HBaseReadExample은 다른 일을하고있다 : 그것은 데이터 집합 (FLINK의 일괄 처리 추상화) 내로 HBase를 테이블을 읽습니다. 사용자 기능이 코드를 사용하면 FLINK 프로그램 내에서 FLINK 프로그램을 시작하는 것을 의미한다.

   사용 사례의 경우, 직접 사용자 기능에 HBase를 클라이언트를 생성하고 상호 작용해야합니다. 이 작업을 수행하는 가장 좋은 방법은 RichFlatMapFunction를 사용하고 open () 메서드에서 HBase와의 연결을 만드는 것입니다.

   FLINK의 다음 버전 (1.2.0)은 크게 응용 프로그램의 처리량을 향상해야 사용자 기능에서 비동기 I / O 작업에 대한 지원 기능을합니다.

from https://stackoverflow.com/questions/40262790/how-to-read-and-write-to-hbase-in-flink-streaming-job by cc-by-sa and MIT license