spark2-shell
import org.apache.hadoop.hbase.HBaseConfiguration
import org.apache.hadoop.hbase.mapreduce.TableInputFormat
import org.apache.hadoop.hbase.client.HBaseAdmin
import org.apache.hadoop.hbase.{HTableDescriptor,HColumnDescriptor}
import org.apache.hadoop.hbase.util.Bytes
import org.apache.hadoop.hbase.client.{Put,HTable}
val tablename = "<tableName>"
val conf = HBaseConfiguration.create()
conf: org.apache.hadoop.conf.Configuration Конфигурация: core-default.xml, core-site.xml, mapred-default.xml, mapred-site.xml, yarn-default.xml, пряжа-site.xml, hdfs-default.xml, hdfs-site.xml, hbase-default.xml, hbase-site.xml
conf.set(TableInputFormat.INPUT_TABLE, tablename)
val admin = new HBaseAdmin(conf)
предупреждение: было одно предупреждение об устаревании;перезапустите с параметром -deprecation для admin: org.apache.hadoop.hbase.client.HBaseAdmin = org.apache.hadoop.hbase.client.HBaseAdmin@1e79b0fc
val hrdd = sc.newAPIHadoopRDD( conf, classOf[TableInputFormat], classOf[org.apache.hadoop.hbase.io.ImmutableBytesWritable], classOf[org.apache.hadoop.hbase.client.Result])
hrdd: org.apache.spark.rdd.RDD [(org.apache.hadoop.hbase.io.ImmutableBytesWritable, org.apache.hadoop.hbase.client.Result)] = NewHadoopRDD [1] в newAPIHadoopRDD по адресу: 32
val rdd = hrdd.map(tuple => tuple._2)
1010* rdd: org.apache.spark.rdd.RDD [org.apache.hadoop.hbase.client.Result] = MapPartitionsRDD [2] на карте по адресу: 31
rdd.take(1)
res3: массив [org.apache.hadoop.hbase.client.Result] = Array (keyvalues = {\ x00 \ x00002250939-1 ^ Main_P \ x00 \ x08 / 0: \ x00 \ x001 \ xA2j \ xBC \ xBE / 1533844953238 / Put / vlen = 187 /seqid = 0, \ x00 \ x00002250939-1 ^ Main_P \ x00 \ x08 / 0: \ x00 \ x002teU \ x9A / 1533844953238 / Put / vlen = 341 / seqid = 0, \ x00 \ x00002250939-1 ^ Main_P \ x00 \ x08/ 0: \ x00 \ x003 | \ x97 \ x0C \ xAB / 1533844953238 / Put / vlen = 123 / seqid = 0, \ x00 \ x00002250939-1 ^ Main_P \ x00 \ x08 / 0: \ x00 \ x005 \ xC0 \ xC9\ x10l / 1533844953238 / Put / vlen = 657 / seqid = 0, \ x00 \ x00002250939-1 ^ Main_P \ x00 \ x08 / 0: \ x00 \ x00: \ xBD \ x9CY_ / 1533844953238 / Put / vlen = 605 /seqid = 0, \ x00 \ x00002250939-1 ^ Main_P \ x00 \ x08 / 0: \ x00 \ x00Ox \ x87 \ xE9 \ xFC / 1533844953238 / Put / vlen = 102 / seqid = 0, \ x00 \ x00002250939-1 ^ Main_P\ x00 \ x08 / 0: \ x00 \ x00f \ x19E \ x9Cn / 1533844953238 / Put / vlen = 89 / seqid = 0, \ x00 \ x00002250939-1 ^ Main_P \ x00 \ x08 / 0: \ x00 \ x00g \ x15 \xE3e) / 15338449532 ...
Эти данные находятся в byteArray
val keyValueRDD = rdd.map(result => (Bytes.toString(result.getRow()).split(",")(0), Bytes.toString(result.value)))
keyValueRDD: org.apache.spark.rdd.RDD [(String, String)] = MapPartitionsRDD [4]на карте в: 31
keyValueRDD.take(1)
res6: Array [(String, String)] = Array ((?? 002250939-1 ^ Main_P ??, �? t {"1005": "2018-06-23 22: 34: 46,0 ?? ("," 1013 ":" 1?? 34 ":" Y?: DCOUNTRY CLUB APARTMENT INC? $ 40 ": []," 1? E07-03 10: 35: 40? E? 683 "," 2?! 08-09 20:02:33?!? 195 "," 3 I? 7? (Z� ?? 8?! T3-31 11: 25: 03.481944 "}))
Выходные данные с перегруженной функцией toString содержат некоторые раздробленные данные и неполные данные STRING в том виде, в котором они хранятся.
Пожалуйста, дайте мне знать, что мне здесь не хватает.
Спасибо,