|
@@ -19,6 +19,9 @@ case class HbaseAsyncFunction() extends RichAsyncFunction[(String, String), (Str
|
|
@transient private var connection: Connection = null
|
|
@transient private var connection: Connection = null
|
|
|
|
|
|
|
|
|
|
|
|
+
|
|
|
|
+
|
|
|
|
+
|
|
override def open(parameters: Configuration): Unit = {
|
|
override def open(parameters: Configuration): Unit = {
|
|
import com.alibaba.ververica.connector.shaded.hadoop3.org.apache.hadoop.conf.Configuration
|
|
import com.alibaba.ververica.connector.shaded.hadoop3.org.apache.hadoop.conf.Configuration
|
|
val configuration: Configuration = HbaseConfig.getHbaseConfiguration()
|
|
val configuration: Configuration = HbaseConfig.getHbaseConfiguration()
|
|
@@ -33,7 +36,7 @@ case class HbaseAsyncFunction() extends RichAsyncFunction[(String, String), (Str
|
|
val table = connection.getTable(TableName.valueOf(tn.toUpperCase())).asInstanceOf[HTable]
|
|
val table = connection.getTable(TableName.valueOf(tn.toUpperCase())).asInstanceOf[HTable]
|
|
val get = new Get(rowkey.getBytes)
|
|
val get = new Get(rowkey.getBytes)
|
|
val result = table.get(get)
|
|
val result = table.get(get)
|
|
- val string = result.toJsonString
|
|
|
|
|
|
+ val string = result.toJson
|
|
resultFuture.complete(Seq((tn, rowkey, string)))
|
|
resultFuture.complete(Seq((tn, rowkey, string)))
|
|
} catch {
|
|
} catch {
|
|
case ex: Exception => {
|
|
case ex: Exception => {
|