Hbase结合MapReduce的批量导入

01

package

Hbase;

02

03

import

java.io.IOException;

04

import

java.text.SimpleDateFormat;

05

import

java.util.Date;

06

07

import

org.apache.hadoop.conf.Configuration;

08

import

org.apache.hadoop.hbase.client.Put;

09

import

org.apache.hadoop.hbase.mapreduce.TableOutputFormat;

10

import

org.apache.hadoop.hbase.mapreduce.TableReducer;

11

import

org.apache.hadoop.hbase.util.Bytes;

12

import

org.apache.hadoop.io.LongWritable;

13

import

org.apache.hadoop.io.NullWritable;

14

import

org.apache.hadoop.io.Text;

15

import

org.apache.hadoop.mapreduce.Counter;

16

import

org.apache.hadoop.mapreduce.Job;

17

import

org.apache.hadoop.mapreduce.Mapper;

18

import

org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

19

import

org.apache.hadoop.mapreduce.lib.input.TextInputFormat;

20

21

public

class

BatchImport {

22

static

class

BatchImportMapper

extends

Mapper<LongWritable,Text,LongWritable,Text>{

23

SimpleDateFormat dateformat1=

new

SimpleDateFormat(

"yyyyMMddHHmmss"

);

24

Text v2 =

new

Text();

25

26

protected

void

map(LongWritable key,Text value,Context context)

throws

java.io.IOException ,InterruptedException {

27

final

String[] splited = value.toString().split(

"\t"

);

28

try

{

29

final

Date date =

new

Date(Long.parseLong(splited[

0

].trim()));

30

final

String dateFormat = dateformat1.format(date);

31

String rowKey = splited[

1

]+

":"

+dateFormat;

32

v2.set(rowKey+

"\t"

+value.toString());

33

context.write(key,v2);

34

}

catch

(NumberFormatException e) {

35

final

Counter counter = context.getCounter(

"BatchImport"

,

"ErrorFormat"

);

36

counter.increment(1L);

37

System.out.println(

"出错了"

+splited[

0

]+

" "

+e.getMessage());

38

}

39

};

40

}

41

42

static

class

BatchImportReducer

extends

TableReducer<LongWritable,Text,NullWritable>{

43

protected

void

reduce(LongWritable key,java.lang.Iterable<Text> values,Context context) 

throws

java.io.IOException ,InterruptedException {

44

for

(Text text : values) {

45

final

String[] splited = text.toString().split(

"\t"

);

46

47

final

Put put =

new

Put(Bytes.toBytes(splited[

0

]));

48

put.add(Bytes.toBytes(

"cf"

),Bytes.toBytes(

"date"

),Bytes.toBytes(splited[

1

]));

49

put.add(Bytes.toBytes(

"cf"

),Bytes.toBytes(

"msisdn"

),Bytes.toBytes(splited[

2

]));

50

put.add(Bytes.toBytes(

"cf"

),Bytes.toBytes(

"apmac"

),Bytes.toBytes(splited[

3

]));

51

//省略其他字段，调用put.add(....)即可

52

context.write(NullWritable.get(),put);

53

}

54

};

55

}

56

57

public

static

void

main(String[] args)

throws

Exception {

58

59

60

final

Configuration configuration =

new

Configuration();

61

//设置zookeeper

62

configuration.set(

"hbase.zookeeper.quorum"

,

"hadoop"

);

63

//设置hbase表名称

64

configuration.set(TableOutputFormat.OUTPUT_TABLE,

"wlan_log"

);

65

//将该值改大，防止hbase超时退出

66

configuration.set(

"dfs.socket.timeout"

,

"180000"

);

67

68

final

Job job =

new

Job(configuration,

"HBaseBatchImport"

);

69

70

job.setMapperClass(BatchImportMapper.

class

);

71

job.setReducerClass(BatchImportReducer.

class

);

72

//设置map的输出，不设置reduce的输出类型

73

job.setMapOutputKeyClass(LongWritable.

class

);

74

job.setMapOutputValueClass(Text.

class

);

75

76

job.setInputFormatClass(TextInputFormat.

class

);

77

//不再设置输出路径，而是设置输出格式类型

78

job.setOutputFormatClass(TableOutputFormat.

class

);

79

80

FileInputFormat.setInputPaths(job,

"hdfs://hadoop:9000/input"

);

81

82

job.waitForCompletion(

true

);

83

}

84

}

01

查询手机

13450456688

的所有上网记录

02

public

static

void

scan(String tableName)

throws

IOException{

03

HTable table =

new

HTable(getConfiguration(),tableName);

04

Scan scan =

new

Scan();

05

scan.setStartRow(Bytes.toBytes(

"13450456688:/"

));

06

scan.setStopRow(Bytes.toBytes(

"13450456688::"

));

07

ResultScanner scanner =table.getScanner(scan);

08

int

i=

0

;

09

for

(Result result : scanner) {

10

System.out.println(

"Scan:"

+i+++

" "

+result);

11

}

12

}

01

查询

134

号段的所有上网记录

02

public

static

void

scanPeriod(String tableName)

throws

IOException{

03

HTable table =

new

HTable(getConfiguration(),tableName);

04

Scan scan =

new

Scan();

05

scan.setStartRow(Bytes.toBytes(

"134/"

));

06

scan.setStopRow(

07

Bytes.toBytes(

"134:"

));

08

scan.setMaxVersions(

1

);

09

ResultScanner scanner =table.getScanner(scan);

10

int

i=

0

;

11

for

(Result result : scanner) {

12

System.out.println(

"Scan:"

+i+++

" "

+result);

13

}

14

}