大数据点点滴滴积少成多---进阶之路_错误: 找不到或无法加载主类 hadoop1.log core file size (blocks,

滨＃可执行文件（Hadoop的的功能操作命令）

等＃配置文件

包括    

LIB＃本地库文件（数据压缩编解码，本地文件系统操作）

的的libexec    

LICENSE.TXT

NOTICE.txt

的的README.txt

sbin目录＃可执行文件（Hadoop的集群进程管理的操作命令）

分享＃开发所需要的JAR包及用户帮助文档

ssh-copy-id hadoop-server03

配置conf =新配置（）

FileSystem fs = FileSystem.get（conf）

dfs.namenode.checkpoint.check.period = 60   ＃检查触发条件是否满足的频率，60秒

dfs.namenode.checkpoint.dir =文件：// $ {} hadoop.tmp.dir / DFS / namesecondary

＃以上两个参数做checkpoint 操作时，secondary namenode 的本地工作目录

dfs.namenode.checkpoint.edits.dir = $ {} dfs.namenode.checkpoint.dir

dfs.namenode.checkpoint.max-retries = 3   ＃最大重试次数

dfs.namenode.checkpoint.period = 3600   ＃两次检查点之间的时间间隔3600秒

dfs.namenode.checkpoint.txns = 1000000  ＃两次checkpoint之间最大的操作记录

fs.trash.checkpoint.interval = 0      ＃回收站过期机制检查频率（分钟）

fs.trash.interval = 0      ＃回收站中文件过期的时间限制（分钟）

// RCP通信的两端共同遵守的协议（本质上就是业务实现类的接口）

public interface ClientNameNodeProtocal {

// RPC通信双方一致的版本号

public static final long versionID = 1L;

//业务方法签名

public String getMetaData（String path）;

}

/ **

 *业务的具体实现类，应该运行在远端服务器上

 * @author duanhaitao@itcast.cn

 *

 * /

公共类NamNodeNameSystemImpl实现ClientNameNodeProtocal {

@覆盖

public String getMetaData（String path）{

//许多逻辑代码用于在元数据池中查找元数据

返回“{/aa/bb/bian4.mp4;300M; [BLK_1，BLK_2，BLK_3]; 3;

{[BLK_1：DN-A，DN-B，DN-E]，[BLK_2：DN-A，DN-B，DN-C]，[BLK_3：DN-A，DN-d，DN-E]}} “;

}

}

/ **

 * RCP服务发布工具

 * @author duanhaitao@itcast.cn

 *

 * /

公共类PublishServiceTool {

public static void main（String [] args）抛出HadoopIllegalArgumentException，IOException {

//创建一个RPC服务建设者

Builder builder = new RPC.Builder（new Configuration（））;

//将要发布的服务的信息设置到建设者中

builder.setBindAddress（“spark01”）。setPort（10000）.setProtocol（ClientNameNodeProtocal.class）.setInstance（new NamNodeNameSystemImpl（））;

//用建设者构建出一个插座服务

Server server = builder.build（）;

//将服务启动，就可以等待客户端请求

server.start（）;

}

}

公共类客户{

public static void main（String [] args）throws Exception {

//首先用RPC框架获得要调用的远端服务的引用（动态代理对象）

ClientNameNodeProtocal namenodeImpl = RPC.getProxy（ClientNameNodeProtocal.class，1L，new InetSocketAddress（“spark01”，10000），new Configuration（））;

//因为这个动态代理对象实现了业务类的接口，所以可以直接通过这个引用来调用业务类的实现方法（本质上，具体实现在远端，走的是插座通信请求）

String metaData = namenodeImpl.getMetaData（“/ aa / bb / bian4.mp4”）;

的的System.out.println（元数据）;

}

}

＃远程调试的NameNode的

export HADOOP_NAMENODE_OPTS =“ - agentlib：jdwp = transport = dt_socket，address = 8888，server = y，suspend = y”

＃远程调试的数据节点

export HADOOP_DATANODE_OPTS =“ - agentlib：jdwp = transport = dt_socket，address = 9888，server = y，suspend = y”

//首先要定义四个泛型的类型

// keyin：LongWritable valuein：Text

// keyout：Text valueout：IntWritable

公共类WordCountMapper扩展Mapper <LongWritable，Text，Text，IntWritable> {

// map方法的生命周期：框架每传一行数据就被调用一次

// key：这一行的起始点在文件中的偏移量

//值：这一行的内容

@覆盖

protected void map（LongWritable key，Text value，Context context）抛出IOException，InterruptedException {

//拿到一行数据转换为字符串

String line = value.toString（）;

//将这一行切分出各个单词

String [] words = line.split（“”）;

//遍历数组，输出<单词，1>

for（String word：words）{

context.write（new Text（word），new IntWritable（1））;

}

}

}

//生命周期：框架每传递进来一个千伏组，减少方法被调用一次

@覆盖

protected void reduce（Text key，Iterable <IntWritable> values，Context context）抛出IOException，InterruptedException {

//定义一个计数器

int count = 0;

//遍历这一组千伏的所有V，累加到计数中

for（IntWritable value：values）{

count + = value.get（）;

}

context.write（key，new IntWritable（count））;

}

}

公共类WordCountRunner {

//把业务逻辑相关的信息（哪个是映射器，哪个是减速，要处理的数据在哪里，输出的结果放哪里......）描述成一个工作对象

//把这个描述好的作业提交给集群去运行

public static void main（String [] args）throws Exception {

配置conf = new Configuration（）;

Job wcjob = Job.getInstance（conf）;

//指定我这个工作所在的JAR包

// wcjob.setJar（“/ home / hadoop / wordcount.jar”）;

wcjob.setJarByClass（WordCountRunner.class）;

wcjob.setMapperClass（WordCountMapper.class）;

wcjob.setReducerClass（WordCountReducer.class）;

//设置我们的业务逻辑映射类的输出键和值的数据类型

wcjob.setMapOutputKeyClass（Text.class）;

wcjob.setMapOutputValueClass（IntWritable.class）;

//设置我们的业务逻辑减速类的输出键和值的数据类型

wcjob.setOutputKeyClass（Text.class）;

wcjob.setOutputValueClass（IntWritable.class）;

//指定要处理的数据所在的位置

FileInputFormat.setInputPaths（wcjob，“hdfs：// hdp-server01：9000 / wordcount / data / big.txt”）;

//指定处理完成之后的结果所保存的位置

FileOutputFormat.setOutputPath（wcjob，new Path（“hdfs：// hdp-server01：9000 / wordcount / output /”））;

//向纱线集群提交这个工作

boolean res = wcjob.waitForCompletion（true）;

System.exit（RES？0：1）;

}

公共类TestSeri {

public static void main（String [] args）throws Exception {

//定义两个ByteArrayOutputStream，用来接收不同序列化机制的序列化结果

ByteArrayOutputStream ba = new ByteArrayOutputStream（）;

ByteArrayOutputStream ba2 = new ByteArrayOutputStream（）;

//定义两个DataOutputStream类类，用于将普通对象进行JDK标准序列化

DataOutputStream dout = new DataOutputStream（ba）;

DataOutputStream dout2 = new DataOutputStream（ba2）;

ObjectOutputStream obout = new ObjectOutputStream（dout2）;

//定义两个豆，作为序列化的源对象

ItemBeanSer itemBeanSer = new ItemBeanSer（1000L，89.9f）;

ItemBean itemBean = new ItemBean（1000L，89.9f）;

//用于比较字符串类型和文本类型的序列化差别

文本atext = new Text（“a”）;

// atext.write（dout）;

itemBean.write（DOUT）;

byte [] byteArray = ba.toByteArray（）;

//比较序列化结果

的的System.out.println（byteArray.length）;

for（byte b：byteArray）{

是System.out.print是（b）中中;

是System.out.print（“：”）;

}

的System.out.println（“-----------------------”）;

String astr =“a”;

// dout2.writeUTF（astr）;

obout.writeObject（itemBeanSer）;

byte [] byteArray2 = ba2.toByteArray（）;

的的System.out.println（byteArray2.length）;

for（byte b：byteArray2）{

是System.out.print是（b）中中;

是System.out.print（“：”）;

}

}

}

/ **

 *全排序示例

 * @author duanhaitao@itcast.cn

 *

 * /

公共类TotalSort {

static class TotalSortMapper扩展Mapper <Text，Text，Text，Text> {

OrderBean bean = new OrderBean（）;

@覆盖

protected void map（文本键，文本值，上下文上下文）抛出IOException，InterruptedException {

// String line = value.toString（）;

// String [] fields = line.split（“\ t”）;

// bean.set（fields [0]，Double.parseDouble（fields [1]））;

context.write（键，值）;

}

}

static class TotalSortReducer扩展Reducer <Text，Text，Text，Text> {

@覆盖

protected void reduce（文本键，Iterable <Text>值，Context上下文）抛出IOException，InterruptedException {

for（Text v：values）{

context.write（键，V）;

}

}

}

public static void main（String [] args）throws Exception {

配置conf = new Configuration（）;

Job job = Job.getInstance（conf）;

job.setJarByClass（TotalSort.class）;

job.setMapperClass（TotalSortMapper.class）;

job.setReducerClass（TotalSortReducer.class）;

// job.setOutputKeyClass（OrderBean.class）;

// job.setOutputValueClass（NullWritable.class）;

//用来读取序列源文件的输入组件

job.setInputFormatClass（SequenceFileInputFormat.class）;

FileInputFormat.setInputPaths（job，new Path（args [0]））;

FileOutputFormat.setOutputPath（job，new Path（args [1]））;

// job.setPartitionerClass（RangePartitioner.class）;

//分区的逻辑使用的Hadoop的的自带的全局排序分区组件

job.setPartitionerClass（TotalOrderPartitioner.class）;

//系统自带的这个抽样器只能针对sequencefile抽样

RandomSampler randomSampler

 = new InputSampler.RandomSampler <Text，Text>（0.1,100,10）;

InputSampler.writePartitionFile（工作，randomSampler）;

//获取抽样器所产生的分区规划描述文件

配置conf2 = job.getConfiguration（）;

String partitionFile = TotalOrderPartitioner.getPartitionFile（conf2）;

//把分区描述规划文件分发到每一个任务节点的本地

job.addCacheFile（new URI（partitionFile））;

//设置若干并发的减少任务

job.setNumReduceTasks（3）;

job.waitForCompletion（真）;

}

}

/ **

 *用于控制洗牌过程中减少端对KV对的聚合逻辑

 * @author duanhaitao@itcast.cn

 *

 * /

公共类ItemidGroupingComparator扩展WritableComparator {

protected ItemidGroupingComparator（）{

超（OrderBean.class，真）;

}

@覆盖

public int compare（WritableComparable a，WritableComparable b）{

OrderBean abean =（OrderBean）a;

OrderBean bbean =（OrderBean）b;

//将ITEM_ID相同的豆都视为相同，从而聚合为一组

return abean.getItemid（）。compareTo（bbean.getItemid（））;

}

}

/ **

 *订单信息豆，实现的Hadoop的序列化机制

 * @author duanhaitao@itcast.cn

 *

 * /

公共类OrderBean实现WritableComparable <OrderBean> {

私人文本的itemid;

私人DoubleWritable金额;

public OrderBean（）{

}

public OrderBean（Text itemid，DoubleWritable amount）{

集（商品ID，金额）;

}

public void set（Text itemid，DoubleWritable amount）{

this.itemid = itemid;

this.amount =金额;

}

public Text getItemid（）{

return itemid;

}

public DoubleWritable getAmount（）{

退货金额;

}

@覆盖

public int compareTo（OrderBean o）{

int cmp = this.itemid.compareTo（o.getItemid（））;

if（cmp == 0）{

cmp = -this.amount.compareTo（o.getAmount（））;

}

返回CMP;

}

@覆盖

public void write（DataOutput out）抛出IOException {

out.writeUTF（itemid.toString（））;

out.writeDouble（amount.get（））;

}

@覆盖

public void readFields（DataInput in）throws IOException {

String readUTF = in.readUTF（）;

double readDouble = in.readDouble（）;

this.itemid = new Text（readUTF）;

this.amount = new DoubleWritable（readDouble）;

}

@覆盖

public String toString（）{

return itemid.toString（）+“\ t”+ amount.get（）;

}

}

公共类ItemIdPartitioner扩展了Partitioner <OrderBean，NullWritable> {

@覆盖

public int getPartition（OrderBean key，NullWritable value，int numPartitions）{

//指定ITEM_ID相同的豆发往相同的减速任务

返回（key.getItemid（）的hashCode（）＆Integer.MAX_VALUE的。）％numPartitions;

}

}

/ **

 *利用secondarysort机制输出每种项目订单金额最大的记录

 * @author duanhaitao@itcast.cn

 *

 * /

公共类SecondarySort {

static class SecondarySortMapper扩展Mapper <LongWritable，Text，OrderBean，NullWritable> {

OrderBean bean = new OrderBean（）;

@覆盖

protected void map（LongWritable key，Text value，Context context）抛出IOException，InterruptedException {

String line = value.toString（）;

String [] fields = StringUtils.split（line，“\ t”）;

bean.set（new Text（fields [0]），new DoubleWritable（Double.parseDouble（fields [1]）））;

context.write（豆，NullWritable.get（））;

}

}

static class SecondarySortReducer扩展Reducer <OrderBean，NullWritable，OrderBean，NullWritable> {

//在设置了groupingcomparator以后，这里收到的kv数据就是：<1001 87.6>，null <1001 76.5>，null ....

//此时，减少方法中的参数key就是上述kv组中的第一个kv的密钥：<1001 87.6>

//要输出同一个项目的所有订单中最大金额的那一个，就只要输出这个关键

@覆盖

protected void reduce（OrderBean key，Iterable <NullWritable> values，Context context）抛出IOException，InterruptedException {

context.write（键，NullWritable.get（））;

}

}

public static void main（String [] args）throws Exception {

配置conf = new Configuration（）;

Job job = Job.getInstance（conf）;

job.setJarByClass（SecondarySort.class）;

job.setMapperClass（SecondarySortMapper.class）;

job.setReducerClass（SecondarySortReducer.class）;

job.setOutputKeyClass（OrderBean.class）;

job.setOutputValueClass（NullWritable.class）;

FileInputFormat.setInputPaths（job，new Path（args [0]））;

FileOutputFormat.setOutputPath（job，new Path（args [1]））;

//指定洗牌所使用的GroupingComparator类

job.setGroupingComparatorClass（ItemidGroupingComparator.class）;

//指定洗牌所使用的分区类

job.setPartitionerClass（ItemIdPartitioner.class）;

job.setNumReduceTasks（3）;

job.waitForCompletion（真）;

}

}

   long minSize = Math.max（getFormatMinSplitSize（），getMinSplitSize（job））;

    long maxSize = getMaxSplitSize（job）;

    public static long getMaxSplitSize（JobContext context）{

returncontext.getConfiguration（）。getLong（SPLIT_MAXSIZE，Long.MAX_VALUE）;

      }

// mapreduce.input.fileinputformat.split.minsize配置这个值可以让切片大小>块大小  

// mapreduce.input.fileinputformat.split.maxsize配置这个值可以让切片大小<块大小

long splitSize = computeSplitSize（blockSize，minSize，maxSize）;

//计算切片大小

protected long computeSplitSize（long blockSize，long minSize，long maxSize）{

    返回Math.max（minSize，Math.min（maxSize，blockSize））;

}

splits.add（makeSplit（路径，长度bytesRemaining，splitSize，blkLocations [blkIndex] .getHosts（）））;

公共类OrderJoin {

static class OrderJoinMapper扩展Mapper <LongWritable，Text，Text，OrderJoinBean> {

@覆盖

protected void map（LongWritable key，Text value，Context context）抛出IOException，InterruptedException {

//拿到一行数据，并且要分辨出这行数据所属的文件

String line = value.toString（）;

String [] fields = line.split（“\ t”）;

//拿到itemid

String itemid = fields [0];

//获取到这一行所在的文件名（通过inpusplit）

String name =“你拿到的文件名”;

//根据文件名，切分出各字段（如果是a，切分出两个字段，如果是b，切分出3个字段）

OrderJoinBean bean = new OrderJoinBean（）;

bean.set（null，null，null，null，null）;

context.write（new Text（itemid），bean）;

}

}

static class OrderJoinReducer extends Reducer <Text，OrderJoinBean，OrderJoinBean，NullWritable> {

@覆盖

protected void reduce（Text key，Iterable <OrderJoinBean> beans，Context context）抛出IOException，InterruptedException {

 //拿到的钥匙是某一个商品ID，比如1000

//拿到的咖啡豆是来自于两类文件的豆

// {1000，金额} {1000，金额} {1000，金额} --- {1000，价格，名称}

//将来自于b文件的豆里面的字段，跟来自于一个的所有豆进行字段拼接并输出

}

}

}

公共类TestDistributedCache {

static class TestDistributedCacheMapper扩展Mapper <LongWritable，Text，Text，Text> {

FileReader in = null;

BufferedReader reader = null;

HashMap <String，String> b_tab = new HashMap <String，String>（）;

String localpath = null;

String uirpath = null;

//是在地图任务初始化的时候调用一次

@覆盖

protected void setup（Context context）抛出IOException，InterruptedException {

//通过这几句代码可以获取到缓存文件的本地绝对路径，测试验证用

Path [] files = context.getLocalCacheFiles（）;

localpath = files [0] .toString（）;

URI [] cacheFiles = context.getCacheFiles（）;

//缓存文件的用法 - 直接用本地IO来读取

//这里读的数据是map task所在机器本地工作目录中的一个小文件

in = new FileReader（“b.txt”）;

reader = new BufferedReader（in）;

String line = null;

而（空！=（线= reader.readLine（）））{

String [] fields = line.split（“，”）;

b_tab.put（字段[0]，字段[1]）;

}

IOUtils.closeStream（读取器）;

IOUtils.closeStream（IN）;

}

@覆盖

protected void map（LongWritable key，Text value，Context context）抛出IOException，InterruptedException {

//这里读的是这个地图任务所负责的那一个切片数据（在hdfs上）

 String [] fields = value.toString（）。split（“\ t”）;

 String a_itemid = fields [0];

 字符串a_amount = fields [1];

 String b_name = b_tab.get（a_itemid）;

 //输出结果1001 98.9 banan

 context.write（new Text（a_itemid），new Text（a_amount +“\ t”+“：”+ localpath +“\ t”+ b_name））;

}

}

public static void main（String [] args）throws Exception {

配置conf = new Configuration（）;

Job job = Job.getInstance（conf）;

job.setJarByClass（TestDistributedCache.class）;

job.setMapperClass（TestDistributedCacheMapper.class）;

job.setOutputKeyClass（Text.class）;

job.setOutputValueClass（LongWritable.class）;

//这里是我们正常的需要处理的数据所在路径

FileInputFormat.setInputPaths（job，new Path（args [0]））;

FileOutputFormat.setOutputPath（job，new Path（args [1]））;

//不需要减速

job.setNumReduceTasks（0）;

//分发一个文件到任务进程的工作目录

job.addCacheFile（new URI（“hdfs：// hadoop-server01：9000 / cachefile / b.txt”））;

//分发一个归档文件到任务进程的工作目录

// job.addArchiveToClassPath（archive）;

//分发的jar包到任务节点的类路径下

// job.addFileToClassPath（jarfile）;

job.waitForCompletion（真）;

}

}

job.addCacheFile（new  URI（“hdfs：// hadoop-server01：9000 / cachefile / b.txt” ））;

//分发一个文件到任务进程的工作目录

job.addCacheFile（new URI（“hdfs：// hadoop-server01：9000 / cachefile / b.txt”））;

//分发一个归档文件到任务进程的工作目录

//job.addArchiveToClassPath(archive）;

//分发的jar包到任务节点的类路径下

//job.addFileToClassPath(jarfile）;

in = new FileReader（“b.txt”）;

reader = new BufferedReader（in）;

String line = reader.readLine（）

  public void initialize（InputSplit genericSplit，

                         TaskAttemptContext context）抛出IOException {

    FileSplit split =（FileSplit）genericSplit;

    配置作业= context.getConfiguration（）;

...... ......... ..

    //打开文件并寻找拆分的开始

    final FileSystem fs = file.getFileSystem（job）;

    fileIn = fs.open（file）;

    CompressionCodec codec = new CompressionCodecFactory（job）.getCodec（file）;

    if（null！= codec）{

    ...... ...... 

//我们总是将第一条记录抛弃（文件第一个分割除外）

//因为我们总是在nextKeyValue（）方法中跨拆分多读了一行（文件最后一个拆除除外）

    if（start！= 0）{

      start + = in.readLine（new Text（），0，maxBytesToConsume（start））;

    }

    this.pos = start;

  }

public boolean nextKeyValue（）throws IOException {

    if（key == null）{

      key = new LongWritable（）;

    }

    key.set（POS）;

    if（value == null）{

      value = new Text（）;

    }

    int newSize = 0;

    //使用<=来多读取一行

    while（getFilePosition（）<= end || in.needAdditionalRecordAfterSplit（））{

      newSize = in.readLine（value，maxLineLength，

          Math.max（maxBytesToConsume（pos），maxLineLength））;

      pos + = newSize;

      if（newSize <maxLineLength）{

        打破;

...。...。

  }

//从机架名称映射到它具有的块列表

HashMap <String，List <OneBlockInfo >> rackToBlocks =

new HashMap <String，List <OneBlockInfo >>（）;

//从块映射到它具有副本的节点

HashMap <OneBlockInfo，String []> blockToNodes =

new HashMap <OneBlockInfo，String []>（）;

//从节点映射到它包含的块列表

HashMap <String，List <OneBlockInfo >> nodeToBlocks =

new HashMap <String，List <OneBlockInfo >>（）;

//填充所有文件的所有块longlongLength = 0;

for（int i = 0; i <paths.length; i ++）{

  files [i] = new OneFileInfo（paths [i]，job，

                             rackToBlocks，blockToNodes，nodeToBlocks，rackToNodes）;

  totLength + = files [i] .getLength（）;

}

//保存当前切片所包含的数据块

    ArrayList <OneBlockInfo> validBlocks = new ArrayList <OneBlockInfo>（）;

    //保存当前切片中的数据块属于哪些节点

    ArrayList <String> nodes = new ArrayList <String>（）;

    //保存当前切片的大小long curSplitSize = 0;

    //处理所有节点并创建本地节点的分割。

    //依据处理每个节点上的数据块for（Iterator <Map.Entry <String，List <OneBlockInfo >>> iter = nodeToBlocks.entrySet（）。iterator（）; iter.hasNext（）;）{

      Map.Entry <String，List <OneBlockInfo >> one = iter.next（）;

      nodes.add（one.getKey（））;

      List <OneBlockInfo> blocksInNode = one.getValue（）;

      //对于每个块，将其复制到validBlocks中。从blockToNodes中删除它，以便在//两个不同的分割中不出现相同的块。

      //依次处理每个数据块，注意blockToNodes变量的作用，它保证了同一数据块不会出现在两个切片中for（OneBlockInfo oneblock：blocksInNode）{

        if（blockToNodes.containsKey（oneblock））{

          validBlocks.add（oneblock）;

          blockToNodes.remove（oneblock）;

          curSplitSize + = oneblock.length; //如果累积的分割大小超过最大值，则创建此分割。

          //如果数据块累积大小大于或等于maxSize，则形成一个切片if（maxSize！= 0 && curSplitSize> = maxSize）{

            //创建一个输入拆分并将其添加到拆分数组addCreatedSplit（job，splits，nodes，validBlocks）;

            curSplitSize = 0;

            validBlocks.clear（）;

          }

        }

      }

      //如果剩下任何块，它们的组合大小是

      //大于minSplitNode，然后将它们组合成一个分割。

      //否则将它们添加回未处理的池中。这有可能

      //稍后他们将与同一机架中的其他块组合在一起。

      //如果剩余数据块大小大于或等于minSizeNode，则将这些数据块构成一个切片;

      //如果剩余数据块大小小于minSizeNode，则将这些数据块归还给blockToNodes，交由后期“同一机架”过程处理if（minSizeNode！= 0 && curSplitSize> = minSizeNode）{

        //创建一个输入拆分并将其添加到拆分数组addCreatedSplit（job，splits，nodes，validBlocks）;

      } else {

        for（OneBlockInfo oneblock：validBlocks）{

          blockToNodes.put（oneblock，oneblock.hosts）;

        }

      }

      validBlocks.clear（）;

      nodes.clear（）;

      curSplitSize = 0;

    }

//如果机架中的块低于指定的最小大小，请保留它们

    //在'溢出'中。所有机架处理完成后，这些溢出

    //块将组合成分裂。

    // overflowBlocks用于保存“同一机架”过程处理之后剩余的数据块

    ArrayList <OneBlockInfo> overflowBlocks = new ArrayList <OneBlockInfo>（）;

    ArrayList <String> racks = new ArrayList <String>（）;

    //一遍又一遍地处理所有机架，直到没有其他工作要做。（blockToNodes.size（）> 0）{

      //在移动到下一个机架之前，为此机架创建一个拆分。

      //为每个人创建一个分割后，回到这个机架

      //剩下的架子

      //一次处理一个机架位置，合并所有可能的块

      //作为一个分割驻留在此机架上。（受最小和最大限制

      //分割大小）。

      //遍历所有机架

      //依次处理每个机架for（Iterator <Map.Entry <String，List <OneBlockInfo >>> iter =

           rackToBlocks.entrySet（）迭代（）; iter.hasNext（）;）{

        Map.Entry <String，List <OneBlockInfo >> one = iter.next（）;

        racks.add（one.getKey（））;

        List <OneBlockInfo> blocks = one.getValue（）;

        //对于每个块，将其复制到validBlocks中。从// blockToNodes中删除它，以便同一个块不会出现在//两个不同的splits中.boolean createdSplit = false; //依次处理该机架的每个数据块for（OneBlockInfo oneblock：blocks）{

          if（blockToNodes.containsKey（oneblock））{

            validBlocks.add（oneblock）;

            blockToNodes.remove（oneblock）;

            curSplitSize + = oneblock.length; //如果累积的分割大小超过最大值，则创建此分割。

            //如果数据块累积大小大于或等于maxSize，则形成一个切片if（maxSize！= 0 && curSplitSize> = maxSize）{

              //创建一个输入拆分并将其添加到拆分数组addCreatedSplit（job，splits，getHosts（racks），validBlocks）;

              createdSplit = true;

              打破;

            }

          }

        }

        //如果我们创建了一个分割，那么只需转到下一个rackif（createdSplit）{

          curSplitSize = 0;

          validBlocks.clear（）;

          racks.clear（）;

          继续;

        }

        if（！validBlocks.isEmpty（））{

          //如果剩余数据块大小大于或等于minSizeRack，则将这些数据块构成一个切片if（minSizeRack！= 0 && curSplitSize> = minSizeRack）{

            //如果指定了最小大小，则创建单个拆分

            //否则，将这些块存储到溢出数据结构addCreatedSplit（job，splits，getHosts（racks），validBlocks）;

          } else {

            //这个机架中有几个块还有待处理。

            //将它们保存在“溢出”阻止列表中。这些将在以后合并。

            //如果剩余数据块大小小于minSizeRack，则将这些数据块加入overflowBlocks overflowBlocks.addAll（validBlocks）;

          }

        }

        curSplitSize = 0;

        validBlocks.clear（）;

        racks.clear（）;

      }

    }

//处理所有溢出块（OneBlockInfo oneblock：overflowBlocks）{

 validBlocks.add（oneblock）; curSplitSize + = oneblock.length; //这可能会导致重新添加现有的机架位置，

//但它应该是ok.for（int i = 0; i <oneblock.racks.length; i ++）{

   racks.add（oneblock.racks [I]）; }

//如果累积的分割大小超过最大值，那么

//创建此拆分。

//如果剩余数据块大小大于或等于maxSize，则将这些数据块构成一个切片if（maxSize！= 0 && curSplitSize> = maxSize）{

//创建一个输入拆分并将其添加到拆分数组addCreatedSplit（作业，分裂，getHosts（机架），validBlocks）; curSplitSize = 0;

   validBlocks.clear（）;

   racks.clear（）;

 }

    }

//处理任何剩余的块，如果有的话。（！validBlocks.isEmpty（））{

      addCreatedSplit（作业，分裂，getHosts（机架），validBlocks）;

    }

static class TextMapperA extends Mapper<LongWritable, Text, Text, LongWritable> {

@Override

protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

String line = value.toString();

String[] words = line.split(" ");

for (String w : words) {

context.write(new Text(w), new LongWritable(1));

}

}

}

static class SequenceMapperB extends Mapper<Text, LongWritable, Text, LongWritable> {

@Override

protected void map(Text key, LongWritable value, Context context) throws IOException, InterruptedException {

context.write(key, value);

}

}

public static void main(String[] args) throws Exception {

Configuration conf = new Configuration();

Job job = Job.getInstance(conf);

job.setJarByClass(WordCount.class);

// 为不同路径的文件指定不同的mapper类及inputformat类

MultipleInputs.addInputPath(job, new Path("c:/wordcount/textdata"), TextInputFormat.class, TextMapperA.class);

MultipleInputs.addInputPath(job, new Path("c:/wordcount/seqdata"), SequenceFileInputFormat.class, SequenceMapperB.class);

job.setReducerClass(SameReducer.class);

job.setMapOutputKeyClass(Text.class);

job.setMapOutputValueClass(LongWritable.class);

FileOutputFormat.setOutputPath(job, new Path("c:/wordcount/multiinouts"));

job.waitForCompletion(true);

}

static class WholeFileInputFormat extends FileInputFormat<NullWritable, BytesWritable> {

@Override

         //改写父类逻辑，总是返回false，从而让

protected boolean isSplitable(JobContext context, Path filename) {

return false;

}

@Override

public RecordReader<NullWritable, BytesWritable> createRecordReader(InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException {

              //返回一个自定义的RecordReader用于读取数据

WholeFileRecordReader reader = new WholeFileRecordReader();

reader.initialize(split, context);

return reader;

}

}

class WholeFileRecordReader extends RecordReader<NullWritable, BytesWritable> {

private FileSplit fileSplit;

private Configuration conf;

//定义一个bytes缓存，用来存储一个小文件的数据内容

private BytesWritable value = new BytesWritable();

private boolean processed = false;

//初始化方法，将传入的文件切片对象和context对象赋值给类成员

@Override

public void initialize(InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException {

fileSplit = (FileSplit) split;

conf = context.getConfiguration();

}

@Override

//核心逻辑，用于从源数据中读取数据并封装为KEY  / VALUE

public boolean nextKeyValue() throws IOException, InterruptedException {

//当前小文件处理过，则processed为true

if (!processed) {

byte[] contents = new byte[(int) fileSplit.getLength()];

Path filePath = fileSplit.getPath();

FileSystem fs = filePath.getFileSystem(conf);

FSDataInputStream in = fs.open(filePath);

IOUtils.readFully(in, contents, 0, contents.length);

value.set(contents, 0, contents.length);

IOUtils.closeStream(in);

processed = true;

return true;

}

//如果当前小文件已经处理过，则返回false，以便调用者跳到下一个文件切片的处理

return false;

}

@Override

//返回一个key

public NullWritable getCurrentKey() throws IOException, InterruptedException {

// TODO Auto-generated method stub

return NullWritable.get();

}

@Override

//返回一个values

public BytesWritable getCurrentValue() throws IOException, InterruptedException {

// TODO Auto-generated method stub

return value;

}

@Override

//用于返回进度信息，读完一个小文件即返回1

public float getProgress() throws IOException, InterruptedException {

// TODO Auto-generated method stub

return processed ? 1.0f : 0.0f;

}

@Override

public void close() throws IOException {

// TODO Auto-generated method stub

}

}

static class CountReducer extends Reducer<Text, LongWritable, Text, LongWritable> {

MultipleOutputs<Text, LongWritable> multipleOutputs = null;

//在初始化方法中构造一个multipleOutputs

protected void setup(Context context) throws IOException, InterruptedException {

multipleOutputs = new MultipleOutputs<Text, LongWritable>(context);

};

@Override

protected void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException {

long count = 0;

for (LongWritable value : values) {

count += value.get();

}

if (key.toString().startsWith("a")) {

//可以通过条件判断将不同内容写入不同文件

multipleOutputs.write(key, new LongWritable(count), "c:/multi/outputa/a");

} else {

multipleOutputs.write(key, new LongWritable(count), "c:/sb/outputb/b");

}

}

//一定要对multipleOutputs进行close，否则内容不会真实写入文件

@Override

protected void cleanup(Context context) throws IOException, InterruptedException {

multipleOutputs.close();

}

}

public class FlowOutputFormat extends FileOutputFormat<Text, NullWritable> {

@Override

public RecordWriter<Text, NullWritable> getRecordWriter(

TaskAttemptContext context) throws IOException,

InterruptedException {

FileSystem fs = FileSystem.get(context.getConfiguration());

Path enhancelog = new Path("hdfs://weekend01:9000/enhance/enhanced.log");

Path tocrawl = new Path("hdfs://weekend01:9000/enhance/tocrawl.log");

//构造两个不同的输出流

FSDataOutputStream enhanceOs = fs.create(enhancelog);

FSDataOutputStream tocrawlOs = fs.create(tocrawl);

//通过构造函数将两个流传给FlowRecordWriter

return new FlowRecordWriter(enhanceOs, tocrawlOs);

}

public static class FlowRecordWriter extends

RecordWriter<Text, NullWritable> {

private FSDataOutputStream enhanceOs;

private FSDataOutputStream tocrawlOs;

public FlowRecordWriter(FSDataOutputStream enhanceOs,

FSDataOutputStream tocrawlOs) {

this.enhanceOs = enhanceOs;

this.tocrawlOs = tocrawlOs;

}

//具体的写出动作在write方法中完成

@Override

public void write(Text key, NullWritable value) throws IOException,

InterruptedException {

String line = key.toString();

if (line.contains("itisok")) {

enhanceOs.write(line.getBytes());

} else {

tocrawlOs.write(line.getBytes());

}

}

@Override

public void close(TaskAttemptContext context) throws IOException,

InterruptedException {

if (enhanceOs != null) {

enhanceOs.close();

}

if (tocrawlOs != null) {

tocrawlOs.close();

}

}

}

}

/**

可以通过运行时加参数来传递参数给conf对象

-D property=value

-conf filename ...

-fs uri  等价于 -D fs.defaultFS=uri

-jt host:port 等价于 -D yarn.resourcemanager.address=host:port

-files file1,file2,

-archives archive1,archive2

-libjars jar1,jar2,...

 *

 * @author duanhaitao@itcast.cn

 *

 */

public class TestToolrunner extends Configured implements Tool {

static {

Configuration.addDefaultResource("hdfs-default.xml");

Configuration.addDefaultResource("hdfs-site.xml");

Configuration.addDefaultResource("core-default.xml");

Configuration.addDefaultResource("core-site.xml");

Configuration.addDefaultResource("mapred-default.xml");

Configuration.addDefaultResource("mapred-site.xml");

Configuration.addDefaultResource("yarn-default.xml");

Configuration.addDefaultResource("yarn-site.xml");

}

@Override

public int run(String[] args) throws Exception {

Configuration conf = getConf();

TreeMap<String, String> treeMap = new TreeMap<String,String>();

for (Entry<String, String> ent : conf) {

treeMap.put(ent.getKey(), ent.getValue());

}

for (Entry<String, String> ent : treeMap.entrySet()) {

System.out.printf("%s=%s\n", ent.getKey(), ent.getValue());

}

return 0;

}

public static void main(String[] args) throws Exception {

ToolRunner.run(new TestToolrunner(), args);

}

}

Job job = Job.getInstance(conf);

FileOutputFormat.setCompressOutput(job, true);

FileOutputFormat.setOutputCompressorClass(job, (Class<? extends CompressionCodec>) Class.forName(""));

conf.setBoolean(Job.MAP_OUTPUT_COMPRESS, true);

conf.setClass(Job.MAP_OUTPUT_COMPRESS_CODEC,

                   GzipCodec.class, CompressionCodec.class);

  public void initialize(InputSplit genericSplit,

                         TaskAttemptContext context) throws IOException {

    FileSplit split = (FileSplit) genericSplit;

    Configuration job = context.getConfiguration();

    this.maxLineLength = job.getInt(MAX_LINE_LENGTH, Integer.MAX_VALUE);

    start = split.getStart();

    end = start + split.getLength();

    final Path file = split.getPath();

    // open the file and seek to the start of the split

    final FileSystem fs = file.getFileSystem(job);

    fileIn = fs.open(file);

    //根据文件后缀名创建相应压缩编码的codec

    CompressionCodec codec = new CompressionCodecFactory(job).getCodec(file);

    if (null!=codec) {

      isCompressedInput = true;

      decompressor = CodecPool.getDecompressor(codec);

  //判断是否属于可切片压缩编码类型

      if (codec instanceof SplittableCompressionCodec) {

        final SplitCompressionInputStream cIn =

          ((SplittableCompressionCodec)codec).createInputStream(

            fileIn, decompressor, start, end,

            SplittableCompressionCodec.READ_MODE.BYBLOCK);

 //如果是可切片压缩编码，则创建一个CompressedSplitLineReader读取压缩数据

        in = new CompressedSplitLineReader(cIn, job,

            this.recordDelimiterBytes);

        start = cIn.getAdjustedStart();

        end = cIn.getAdjustedEnd();

        filePosition = cIn;

      } else {

//如果是不可切片压缩编码，则创建一个SplitLineReader读取压缩数据，并将文件输入流转换成解压数据流传递给普通SplitLineReader读取

        in = new SplitLineReader(codec.createInputStream(fileIn,

            decompressor), job, this.recordDelimiterBytes);

        filePosition = fileIn;

      }

    } else {

      fileIn.seek(start);

   //如果不是压缩文件，则创建普通SplitLineReader读取数据

      in = new SplitLineReader(fileIn, job, this.recordDelimiterBytes);

      filePosition = fileIn;

    }

public class MultiOutputs {

//通过枚举形式定义自定义计数器

enum MyCounter{MALFORORMED,NORMAL}

static class CommaMapper extends Mapper<LongWritable, Text, Text, LongWritable> {

@Override

protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

String[] words = value.toString().split(",");

for (String word : words) {

context.write(new Text(word), new LongWritable(1));

}

//对枚举定义的自定义计数器加1

context.getCounter(MyCounter.MALFORORMED).increment(1);

//通过动态设置自定义计数器加1

context.getCounter("counterGroupa", "countera").increment(1);

}

}

      ControlledJob cJob1 = new ControlledJob(job1.getConfiguration());

        ControlledJob cJob2 = new ControlledJob(job2.getConfiguration());

        ControlledJob cJob3 = new ControlledJob(job3.getConfiguration());

        // 设置作业依赖关系

        cJob2.addDependingJob(cJob1);

        cJob3.addDependingJob(cJob2);

        JobControl jobControl = new JobControl("RecommendationJob");

        jobControl.addJob(cJob1);

        jobControl.addJob(cJob2);

        jobControl.addJob(cJob3);

        cJob1.setJob(job1);

        cJob2.setJob(job2);

        cJob3.setJob(job3);

        // 新建一个线程来运行已加入JobControl中的作业，开始进程并等待结束

        Thread jobControlThread = new Thread(jobControl);

        jobControlThread.start();

        while (!jobControl.allFinished()) {

            Thread.sleep(500);

        }

        jobControl.stop();

        return 0;

2012-3-1 a

2012-3-2 b

2012-3-3 c

2012-3-4 d

2012-3-5 a

2012-3-6 b

2012-3-7 c

2012-3-3 c

2

32

654

32

15

756

65223

5956

22

650

92

1    2

2    6

3    15

4    22

5    26

6    32

7    32

8    54

9    92

10    650

11    654

12    756

13    5956

14    65223

1）math：

张三    88

李四    99

王五    66

赵六    77

2）chinese：

张三    78

李四    89

王五    96

赵六    67

3）english：

张三    80

李四    82

王五    84

赵六    86

张三    82

李四    90

王五    82

赵六    76

child parent

Tom Lucy

Tom Jack

Jone Lucy

Jone Jack

Lucy Mary

Lucy Ben

Jack Alice

Jack Jesse

Terry Alice

Terry Jesse

Philip Terry

Philip Alma

Mark Terry

Mark Alma

grandchild grandparent

Tom Alice

Tom Jesse

Jone Alice

Jone Jesse

Tom Mary

Tom Ben

Jone Mary

Jone Ben

Philip Alice

Philip Jesse

Mark Alice

Mark Jesse

A:B,C,D,F,E,O

B:A,C,E,K

C:F,A,D,I

D:A,E,F,L

E:B,C,D,M,L

F:A,B,C,D,E,O,M

G:A,C,D,E,F

H:A,C,D,E,O

I:A,O

J:B,O

K:A,C,D

L:D,E,F

M:E,F,G

O:A,H,I,J

……

A-B C,E,

A-C D,F,

A-D E,F,

A-E B,C,D,

A-F B,C,D,E,O,

A-G C,D,E,F,

A-H C,D,E,O,

A-I O,

A-J B,O,

A-K C,D,

A-L D,E,F,

A-M E,F,

B-C A,

B-D A,E,

……

20140510 09:17:19 uuid-01 search-dancheng dep=北京&arr=上海&date=20140529&pnvm=0

20140510 09:18:20 uuid-02 search-wangFan dep=北京&arr=上海&sdate=20140529&edate=20140605

20140510 09:18:23 uuid-01 detail dep=北京&arr=上海&date=20140529&fcode=CA1810

20140510 09:20:29 uuid-02 detail dep=北京&arr=上海&date=20140529&fcode=CA1810

20140510 09:21:19 uuid-01 submit dep=北京&arr=上海&date=20140529&fcode=CA1810&price=1280

20140510 09:23:19 uuid-03 search-dancheng dep=北京&arr=广州&date=20140529&pnvm=0

20140510 09:25:19 uuid-04 search-dancheng dep=北京&arr西安&date=20140529&pnvm=0

20140510 09:25:30 uuid-05 search-dancheng dep=北京&arr=天津&date=20140529&pnvm=0

20140510 09:26:29 uuid-04 detail dep=北京&arr=西安&上海&date=20140529&fcode=CA1810

20140510 09:28:19 uuid-06 submit dep=北京&arr=拉萨&date=20140529&fcode=CA1810&price=2260

15-06-10.23:58:02.321 [pool-22-thread-5] INFO  HttpPostMessageSender   -  HttpPostMessageSender resp statusCode: 200 content:success tradeno:2015061010001000070650683

15-06-10.23:58:02.321 [pool-22-thread-5] INFO  HttpPostMessageSender   - HttpPost 是否发送成功  true

15-06-10.23:58:02.321 [pool-22-thread-5] INFO  NotificationServiceImpl  - ****进行入库操作****

15-06-10.23:58:02.324 [pool-22-thread-5] INFO  NotificationServiceImpl  - ****没有此TRADE_NO,新增Notification****tradeNo=2015061010001000070650683

15-06-10.23:58:02.327 [pool-22-thread-5] INFO  NotifyServiceImpl       - ****enter--saveOrUpdateNotity****

15-06-10.23:58:02.330 [pool-22-thread-5] INFO  NotifyServiceImpl       - ****没有此TRADE_NO,新增Notify****tradeNo=2015061010001000070650683

15-06-10.23:58:02.333 [pool-22-thread-5] INFO  ACCESS                  - 2015061010001000070650683,FINISHED_SUCCESS

15-06-10.23:58:04.250 [pool-20-thread-2] INFO  NPPListener             - Received a new message OutTradeNotify{tradeInfo=TradeInfo{outTradeNo='150610263916206067998', tradeNo='2015061010001000070650827', originalTradeNo='null', bizTradeNo='9488771051', tradeType=TRADE_GENERAL, subTradeType=SALE, payMethod=CASHIERGATEMODE, tradeMoney=Money{currency=CNY, amount=10420}, tradeSubject='消费订单', submitter=ThinCustomer{merchantNo='23077370', customerNo='360080000230773708', customerLoginName='null', customerName='null', customerOutName='null'}, seller=ThinCustomer{merchantNo='23077370', customerNo='360080000230773708', customerLoginName='null', customerName='null', customerOutName='null'}, sellerAccountNo='360080000230773708000811', buyer=ThinCustomer{merchantNo='null', customerNo='360000000260175680', customerLoginName='null', customerName='null', customerOutName='null'}, tradeStatus=TRADE_FINISHED, createdDate=Wed Jun 10 23:57:32 CST 2015, deadlineTime=null, tradeFinishedDate='20150610', tradeFinishedTime='235802', payTool=EXPRESS, bankCode='CEB', exchangeDate='null', exchangeRate='null', returnParams='null', oldGWV60AuthCode='null', oldEXV10TerminalNo='null', clearingCurrency=null, clearingMoney=null, tradeExtInfo=TradeExtInfo{notifyStatus='NOT', outMessageId='null', cardSha1='null', signNo='null', returnParams='null', extendParams='null', pageBackUrl='null', serverNotifyUrl='http://gw.jd.com/payment/notify_chinabankReal.action', notifySmsMoible='null', notifyMailAddress='null', innerMessageFormat='XML', apiMessageFormat='EX_V1.0', requestCharset='UTF-8', encryptType='3DES', signType='MD5', requestModule='null', requestVersion='null', remoteIp='109.145.60.24', receivingChannel='JDSC', requestProtocol='HTTP', requestMethod='null', outTradeDate='20150610', outTradeTime='235731', outTradeIp='109.145.60.24', outRefererHosts='null', retryCount=1}, ext=null}}, OutMessageNotify{apiMessageFormat=null, messageFormat=null, notifyCharset='null', signType='null', encryptType='null'}, MessageNotify{responseModule='null', responseCode='null', responseDesc='null'}

15-06-10.23:58:04.253 [pool-20-thread-2] INFO  KeyServiceImpl          - Calling SecurityService to get {} key for merchant {} with codeClass {}23077370KeyTypeEnum{code='3DES', cnName='三DES'}EXPRESS

15-06-10.23:58:04.264 [pool-20-thread-2] INFO  CustomerCenterFacade    - [INVOCATION_LOG_C] 2015-06-10.23:58:04.264;pool-20-thread-2;172.17.92.48:0->172.17.87.47:20996;com.wangyin.customer.api.CustomerCenterFacade:1.1.6.getMerchantCustomerKeys(com.wangyin.customer.common.dto.customer.CustomerParamDTO);***;2015-06-10.23:58:04.253;RESULT:***;11,112,359;

15-06-10.23:58:04.264 [pool-20-thread-2] INFO  KeyServiceImpl          - 获取的3DES 密钥值为20B0984A9B751F0B911A1AEA0738D557AE16548CCE029E2A

15-06-10.23:58:04.264 [pool-20-thread-2] INFO  KeyServiceImpl          - Calling SecurityService to get {} key for merchant {} with codeClass {}23077370KeyTypeEnum{code='SALT', cnName='签名密钥'}EXPRESS

15-06-10.23:58:04.270 [pool-24-thread-4] INFO  NPPListener             - Received a new message OutTradeNotify{tradeInfo=TradeInfo{outTradeNo='22015061023575751670871914', tradeNo='2015061010001000070651406', originalTradeNo='null', bizTradeNo='null', tradeType=TRADE_GENERAL, subTradeType=SALE, payMethod=APIEXPRESSMODE, tradeMoney=Money{currency=CNY, amount=500000}, tradeSubject='消费订单', submitter=ThinCustomer{merchantNo='22843776', customerNo='360080000228437761', customerLoginName='null', customerName='null', customerOutName='null'}, seller=ThinCustomer{merchantNo='22843776', customerNo='360080000228437761', customerLoginName='null', customerName='null', customerOutName='null'}, sellerAccountNo='360080000228437761000811', buyer=null, tradeStatus=TRADE_FINISHED, createdDate=Wed Jun 10 23:57:57 CST 2015, deadlineTime=null, tradeFinishedDate='20150610', tradeFinishedTime='235802', payTool=EXPRESS, bankCode='ICBC', exchangeDate='null', exchangeRate='null', returnParams='22894010', oldGWV60AuthCode='null', oldEXV10TerminalNo='00000002', clearingCurrency=null, clearingMoney=null, tradeExtInfo=TradeExtInfo{notifyStatus='NOT', outMessageId='API.150610.0ddf8c2f7ed94f3e9f741cd44500a866', cardSha1='5D72C7755A82576EE906BAB8314164ABAC513C9C', signNo='201505110010089270009113541', returnParams='22894010', extendParams='null', pageBackUrl='null', serverNotifyUrl='http://jrb-api.d.chinabank.com.cn/notify/quick.htm', notifySmsMoible='null', notifyMailAddress='null', innerMessageFormat='XML', apiMessageFormat='EX_V1.0', requestCharset='UTF-8', encryptType='3DES', signType='MD5', requestModule='null', requestVersion='null', remoteIp='172.17.80.168', receivingChannel='API', requestProtocol='HTTP', requestMethod='POST', outTradeDate='null', outTradeTime='null', outTradeIp='null', outRefererHosts='null', retryCount=1}, ext=null}}, OutMessageNotify{apiMessageFormat=null, messageFormat=null, notifyCharset='null', signType='null', encryptType='null'}, MessageNotify{responseModule='null', responseCode='null', responseDesc='null'}

15-06-10.23:58:04.271 [pool-20-thread-2] INFO  CustomerCenterFacade    - [INVOCATION_LOG_C] 2015-06-10.23:58:04.271;pool-20-thread-2;172.17.92.48:0->172.17.91.104:20996;com.wangyin.customer.api.CustomerCenterFacade:1.1.6.getMerchantCustomerKeys(com.wangyin.customer.common.dto.customer.CustomerParamDTO);***;2015-06-10.23:58:04.264;RESULT:***;6,971,110;

15-06-10.23:58:04.272 [pool-20-thread-2] INFO  KeyServiceImpl          - 获取MD5 TOKEN 密钥的值为1qaz2wsx3edc

15-06-10.23:58:04.273 [pool-20-thread-2] INFO  NPPNotifyProcessorImpl  - ApiMessageFormatEX_V1.0

15-06-10.23:58:04.273 [pool-24-thread-4] INFO  KeyServiceImpl          - Calling SecurityService to get {} key for merchant {} with codeClass {}22843776KeyTypeEnum{code='3DES', cnName='三DES'}EXPRESS

15-06-10.23:58:04.273 [pool-20-thread-2] INFO  NPPNotifyProcessorImpl  - 转化为NotificationDTO的结果为: com.wangyin.npp.notify.facade.dto.NotificationDTO@54b27890

15-06-10.23:58:04.273 [pool-20-thread-2] INFO  NotificationServiceImpl  - 准备入库(可能会入库)的 notification=Notification [TRADE_NO=2015061010001000070650827, SOURCE_NAME=NPP_PAYMENT_COMPLETE, FROM_ADDRESS=EXPRESS, FROM_NAME=23077370, TO_ADDRESS=http://gw.jd.com/payment/notify_chinabankReal.action, CHANNEL=HTTP_POST, SUBJECT=null, CONTENT=resp=PD94bWwgdmVyc2lvbj0iMS4wIiBlbmNvZGluZz0iVVRGLTgiPz4NCjxDSElOQUJBTks%2BCiAgPFZFUlNJT04%2BMS4wLjA8L1ZFUlNJT04%2BCiAgPE1FUkNIQU5UPjIzMDc3MzcwPC9NRVJDSEFOVD4KICA8VEVSTUlOQUw%2BMDAwMDAwMDE8L1RFUk1JTkFMPgogIDxEQVRBPllFbm10T0Zkb0RBK0tHdmhVZmJBZTlKOVZDOC9ONGx1YW5uMlBTRFF0L0VNTUh3eHR6L29tYi9vdlArTjAybnlsTGdhbUhCVDBZYVpBMUxoSC9iV3RndmoxN0JMTDhPTFc3U3laZmMxMU5sczRqSFdGeUR1UHNsb3F4YU51aFdUUFFDTzljMCtrTFpDZkpuZHB6d2sxN3J4dU5mRGVuYmljZ21kWHphSlhQNElQZzFKQ2h1ZGRRNWdTQTQ4UWVPVEE0UUhJYUsyQVFJNTNZQU03RHdQWFBrZkNPMythRUgvMk5oeGJMRmtYMTEvalJWUUI0NDM1K2FtSm1zclE0UFJ5cVVSWmx6eGVJQk5XNU4xZnZjMUE1NXRVa1RmRjNWc1orWjU2WkdydFoyQzdnQ3BWNkxqOUNDUWlzbjhKMEd3Z2JLS0kvdUMyUVNDTHJOMUl3YU8waSsxUUFIVWdPRGRtTFZHUGxhSTBqTS85UWVmY0Q2R0FjaVJua214R

yarn.scheduler.fair.allocation.file

Path to allocation file. An allocation file is an XML manifest describing queues and their properties, in addition to certain policy defaults. This file must be in the XML format described in the next section. If a relative path is given, the file is searched for on the classpath (which typically includes the Hadoop conf directory). Defaults to fair-scheduler.xml.

yarn.scheduler.fair.user-as-default-queue

Whether to use the username associated with the allocation as the default queue name, in the event that a queue name is not specified. If this is set to “false” or unset, all jobs have a shared default queue, named “default”. Defaults to true. If a queue placement policy is given in the allocations file, this property is ignored.

yarn.scheduler.fair.preemption

Whether to use preemption. Defaults to false.

yarn.scheduler.fair.preemption.cluster-utilization-threshold

The utilization threshold after which preemption kicks in. The utilization is computed as the maximum ratio of usage to capacity among all resources. Defaults to 0.8f.

yarn.scheduler.fair.sizebasedweight

Whether to assign shares to individual apps based on their size, rather than providing an equal share to all apps regardless of size. When set to true, apps are weighted by the natural logarithm of one plus the app’s total requested memory, divided by the natural logarithm of 2. Defaults to false.

yarn.scheduler.fair.assignmultiple

Whether to allow multiple container assignments in one heartbeat. Defaults to false.

yarn.scheduler.fair.max.assign

If assignmultiple is true, the maximum amount of containers that can be assigned in one heartbeat. Defaults to -1, which sets no limit.

yarn.scheduler.fair.locality.threshold.node

For applications that request containers on particular nodes, the number of scheduling opportunities since the last container assignment to wait before accepting a placement on another node. Expressed as a float between 0 and 1, which, as a fraction of the cluster size, is the number of scheduling opportunities to pass up. The default value of -1.0 means don’t pass up any scheduling opportunities.

yarn.scheduler.fair.locality.threshold.rack

For applications that request containers on particular racks, the number of scheduling opportunities since the last container assignment to wait before accepting a placement on another rack. Expressed as a float between 0 and 1, which, as a fraction of the cluster size, is the number of scheduling opportunities to pass up. The default value of -1.0 means don’t pass up any scheduling opportunities.

yarn.scheduler.fair.allow-undeclared-pools

If this is true, new queues can be created at application submission time, whether because they are specified as the application’s queue by the submitter or because they are placed there by the user-as-default-queue property. If this is false, any time an app would be placed in a queue that is not specified in the allocations file, it is placed in the “default” queue instead. Defaults to true. If a queue placement policy is given in the allocations file, this property is ignored.

3.1添加一个zoo.cfg配置文件

$ZOOKEEPER/conf

mv zoo_sample.cfg zoo.cfg

3.2修改配置文件（zoo.cfg）

dataDir=/itcast/zookeeper-3.4.5/data

server.1=itcast05:2888:3888

server.2=itcast06:2888:3888

server.3=itcast07:2888:3888

3.3在（dataDir=/itcast/zookeeper-3.4.5/data）创建一个myid文件，里面内容是server.N中的N（server.2里面内容为2）

echo "1" > myid

3.4将配置好的zk拷贝到其他节点

scp -r /itcast/zookeeper-3.4.5/ itcast06:/itcast/

scp -r /itcast/zookeeper-3.4.5/ itcast07:/itcast/

3.5注意：在其他节点上一定要修改myid的内容

在itcast06应该讲myid的内容改为2 （echo "6" > myid）

在itcast07应该讲myid的内容改为3 （echo "7" > myid）

4.启动集群

分别启动zk

./zkServer.sh start

conf

输出相关服务配置的详细信息。

cons

列出所有连接到服务器的客户端的完全的连接 / 会话的详细信息。包括“接受 / 发送”的包数量、会话 id 、操作延迟、最后的操作执行等等信息。

dump

列出未经处理的会话和临时节点。

envi

输出关于服务环境的详细信息（区别于 conf 命令）。

reqs

列出未经处理的请求

ruok

测试服务是否处于正确状态。如果确实如此，那么服务返回“imok ”，否则不做任何相应。

stat

输出关于性能和连接的客户端的列表。

wchs

列出服务器 watch 的详细信息。

wchc

通过 session 列出服务器 watch 的详细信息，它的输出是一个与watch 相关的会话的列表。

wchp

通过路径列出服务器 watch 的详细信息。它输出一个与 session相关的路径。

功能

描述

create

在本地目录树中创建一个节点

delete

删除一个节点

exists

测试本地是否存在目标节点

get/set data

从目标节点上读取 / 写数据

get/set ACL

获取 / 设置目标节点访问控制列表信息

get children

检索一个子节点上的列表

sync

等待要被传送的数据

public class SimpleDemo {

// 会话超时时间，设置为与系统默认时间一致

private static final int SESSION_TIMEOUT = 30000;

// 创建 ZooKeeper 实例

ZooKeeper zk;

// 创建 Watcher 实例

Watcher wh = new Watcher() {

public void process(org.apache.zookeeper.WatchedEvent event)

{

System.out.println(event.toString());

}

};

// 初始化 ZooKeeper 实例

private void createZKInstance() throws IOException

{

zk = new ZooKeeper("weekend01:2181", SimpleDemo.SESSION_TIMEOUT, this.wh);

}

private void ZKOperations() throws IOException, InterruptedException, KeeperException

{

System.out.println("/n1. 创建 ZooKeeper 节点 (znode ： zoo2, 数据： myData2 ，权限： OPEN_ACL_UNSAFE ，节点类型： Persistent");

zk.create("/zoo2", "myData2".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

System.out.println("/n2. 查看是否创建成功： ");

System.out.println(new String(zk.getData("/zoo2", false, null)));

System.out.println("/n3. 修改节点数据 ");

zk.setData("/zoo2", "shenlan211314".getBytes(), -1);

System.out.println("/n4. 查看是否修改成功： ");

System.out.println(new String(zk.getData("/zoo2", false, null)));

System.out.println("/n5. 删除节点 ");

zk.delete("/zoo2", -1);

System.out.println("/n6. 查看节点是否被删除： ");

System.out.println(" 节点状态： [" + zk.exists("/zoo2", false) + "]");

}

private void ZKClose() throws InterruptedException

{

zk.close();

}

public static void main(String[] args) throws IOException, InterruptedException, KeeperException {

SimpleDemo dm = new SimpleDemo();

dm.createZKInstance();

dm.ZKOperations();

dm.ZKClose();

}

}

public class AppClient {

private String groupNode = "sgroup";

private ZooKeeper zk;

private Stat stat = new Stat();

private volatile List<String> serverList;

/**

 * 连接zookeeper

 */

public void connectZookeeper() throws Exception {

zk

= new ZooKeeper("localhost:4180,localhost:4181,localhost:4182", 5000, new Watcher() {

public void process(WatchedEvent event) {

// 如果发生了"/sgroup"节点下的子节点变化事件, 更新server列表, 并重新注册监听

if (event.getType() == EventType.NodeChildrenChanged

&& ("/" + groupNode).equals(event.getPath())) {

try {

updateServerList();

} catch (Exception e) {

e.printStackTrace();

}

}

}

});

updateServerList();

}

/**

 * 更新server列表

 */

private void updateServerList() throws Exception {

List<String> newServerList = new ArrayList<String>();

// 获取并监听groupNode的子节点变化

// watch参数为true, 表示监听子节点变化事件.

// 每次都需要重新注册监听, 因为一次注册, 只能监听一次事件, 如果还想继续保持监听, 必须重新注册

List<String> subList = zk.getChildren("/" + groupNode, true);

for (String subNode : subList) {

// 获取每个子节点下关联的server地址

byte[] data = zk.getData("/" + groupNode + "/" + subNode, false, stat);

newServerList.add(new String(data, "utf-8"));

}

// 替换server列表

serverList = newServerList;

System.out.println("server list updated: " + serverList);

}

/**

 * client的工作逻辑写在这个方法中

 * 此处不做任何处理, 只让client sleep

 */

public void handle() throws InterruptedException {

Thread.sleep(Long.MAX_VALUE);

}

public static void main(String[] args) throws Exception {

AppClient ac = new AppClient();

ac.connectZookeeper();

ac.handle();

}

}

public class AppServer {

private String groupNode = "sgroup";

private String subNode = "sub";

/**

 * 连接zookeeper

 * @param address server的地址

 */

public void connectZookeeper(String address) throws Exception {

ZooKeeper zk = new ZooKeeper(

"localhost:4180,localhost:4181,localhost:4182",

5000, new Watcher() {

public void process(WatchedEvent event) {

// 不做处理

}

});

// 在"/sgroup"下创建子节点

// 子节点的类型设置为EPHEMERAL_SEQUENTIAL, 表明这是一个临时节点, 且在子节点的名称后面加上一串数字后缀

// 将server的地址数据关联到新创建的子节点上

String createdPath = zk.create("/" + groupNode + "/" + subNode, address.getBytes("utf-8"),

Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);

System.out.println("create: " + createdPath);

}

/**

 * server的工作逻辑写在这个方法中

 * 此处不做任何处理, 只让server sleep

 */

public void handle() throws InterruptedException {

Thread.sleep(Long.MAX_VALUE);

}

public static void main(String[] args) throws Exception {

// 在参数中指定server的地址

if (args.length == 0) {

System.err.println("The first argument must be server address");

System.exit(1);

}

AppServer as = new AppServer();

as.connectZookeeper(args[0]);

as.handle();

}

}

public class DistributedClient {

    // 超时时间

    private static final int SESSION_TIMEOUT = 5000;

    // zookeeper server列表

    private String hosts = "localhost:4180,localhost:4181,localhost:4182";

    private String groupNode = "locks";

    private String subNode = "sub";

    private ZooKeeper zk;

    // 当前client创建的子节点

    private String thisPath;

    // 当前client等待的子节点

    private String waitPath;

    private CountDownLatch latch = new CountDownLatch(1);

    /**

     * 连接zookeeper

     */

    public void connectZookeeper() throws Exception {

        zk = new ZooKeeper(hosts, SESSION_TIMEOUT, new Watcher() {

            public void process(WatchedEvent event) {

                try {

                    // 连接建立时, 打开latch, 唤醒wait在该latch上的线程

                    if (event.getState() == KeeperState.SyncConnected) {

                        latch.countDown();

                    }

                    // 发生了waitPath的删除事件

                    if (event.getType() == EventType.NodeDeleted && event.getPath().equals(waitPath)) {

                        doSomething();

                    }

                } catch (Exception e) {

                    e.printStackTrace();

                }

            }

        });

        // 等待连接建立

        latch.await();

        // 创建子节点

        thisPath = zk.create("/" + groupNode + "/" + subNode, null, Ids.OPEN_ACL_UNSAFE,

                CreateMode.EPHEMERAL_SEQUENTIAL);

        // wait一小会, 让结果更清晰一些

        Thread.sleep(10);

        // 注意, 没有必要监听"/locks"的子节点的变化情况

        List<String> childrenNodes = zk.getChildren("/" + groupNode, false);

        // 列表中只有一个子节点, 那肯定就是thisPath, 说明client获得锁

        if (childrenNodes.size() == 1) {

            doSomething();

        } else {

            String thisNode = thisPath.substring(("/" + groupNode + "/").length());

            // 排序

            Collections.sort(childrenNodes);

            int index = childrenNodes.indexOf(thisNode);

            if (index == -1) {

                // never happened

            } else if (index == 0) {

                // inddx == 0, 说明thisNode在列表中最小, 当前client获得锁

                doSomething();

            } else {

                // 获得排名比thisPath前1位的节点

                this.waitPath = "/" + groupNode + "/" + childrenNodes.get(index - 1);

                // 在waitPath上注册监听器, 当waitPath被删除时, zookeeper会回调监听器的process方法

                zk.getData(waitPath, true, new Stat());

            }

        }

    }

    private void doSomething() throws Exception {

        try {

            System.out.println("gain lock: " + thisPath);

            Thread.sleep(2000);

            // do something

        } finally {

            System.out.println("finished: " + thisPath);

            // 将thisPath删除, 监听thisPath的client将获得通知

            // 相当于释放锁

            zk.delete(this.thisPath, -1);

        }

    }

    public static void main(String[] args) throws Exception {

        for (int i = 0; i < 10; i++) {

            new Thread() {

                public void run() {

                    try {

                        DistributedClient dl = new DistributedClient();

                        dl.connectZookeeper();

                    } catch (Exception e) {

                        e.printStackTrace();

                    }

                }

            }.start();

        }

        Thread.sleep(Long.MAX_VALUE);

    }

}

public class DistributedClientMy {

// 超时时间

private static final int SESSION_TIMEOUT = 5000;

// zookeeper server列表

private String hosts = "spark01:2181,spark02:2181,spark03:2181";

private String groupNode = "locks";

private String subNode = "sub";

private boolean haveLock = false;

private ZooKeeper zk;

// 当前client创建的子节点

private volatile String thisPath;

/**

 * 连接zookeeper

 */

public void connectZookeeper() throws Exception {

zk = new ZooKeeper("spark01:2181", SESSION_TIMEOUT, new Watcher() {

public void process(WatchedEvent event) {

try {

// 子节点发生变化

if (event.getType() == EventType.NodeChildrenChanged && event.getPath().equals("/" + groupNode)) {

// thisPath是否是列表中的最小节点

List<String> childrenNodes = zk.getChildren("/" + groupNode, true);

String thisNode = thisPath.substring(("/" + groupNode + "/").length());

// 排序

Collections.sort(childrenNodes);

if (childrenNodes.indexOf(thisNode) == 0) {

doSomething();

thisPath = zk.create("/" + groupNode + "/" + subNode, null, Ids.OPEN_ACL_UNSAFE,

CreateMode.EPHEMERAL_SEQUENTIAL);

}

}

} catch (Exception e) {

e.printStackTrace();

}

}

});

// 创建子节点

thisPath = zk.create("/" + groupNode + "/" + subNode, null, Ids.OPEN_ACL_UNSAFE,

CreateMode.EPHEMERAL_SEQUENTIAL);

// wait一小会, 让结果更清晰一些

Thread.sleep(new Random().nextInt(1000));

// 监听子节点的变化

List<String> childrenNodes = zk.getChildren("/" + groupNode, true);

// 列表中只有一个子节点, 那肯定就是thisPath, 说明client获得锁

if (childrenNodes.size() == 1) {

doSomething();

thisPath = zk.create("/" + groupNode + "/" + subNode, null, Ids.OPEN_ACL_UNSAFE,

CreateMode.EPHEMERAL_SEQUENTIAL);

}

}

/**

 * 共享资源的访问逻辑写在这个方法中

 */

private void doSomething() throws Exception {

try {

System.out.println("gain lock: " + thisPath);

Thread.sleep(2000);

// do something

} finally {

System.out.println("finished: " + thisPath);

// 将thisPath删除, 监听thisPath的client将获得通知

// 相当于释放锁

zk.delete(this.thisPath, -1);

}

}

public static void main(String[] args) throws Exception {

DistributedClientMy dl = new DistributedClientMy();

dl.connectZookeeper();

Thread.sleep(Long.MAX_VALUE);

}

}

<configuration>

<!-- 指定hdfs的nameservice为ns1 -->

<property>

<name>fs.defaultFS</name>

<value>hdfs://ns1/</value>

</property>

<!-- 指定hadoop临时目录 -->

<property>

<name>hadoop.tmp.dir</name>

<value>/home/hadoop/app/hadoop-2.4.1/tmp</value>

</property>

<!-- 指定zookeeper地址 -->

<property>

<name>ha.zookeeper.quorum</name>

<value>weekend05:2181,weekend06:2181,weekend07:2181</value>

</property>

</configuration>

configuration>

<!--指定hdfs的nameservice为ns1，需要和core-site.xml中的保持一致 -->

<property>

<name>dfs.nameservices</name>

<value>ns1</value>

</property>

<!-- ns1下面有两个NameNode，分别是nn1，nn2 -->

<property>

<name>dfs.ha.namenodes.ns1</name>

<value>nn1,nn2</value>

</property>

<!-- nn1的RPC通信地址 -->

<property>

<name>dfs.namenode.rpc-address.ns1.nn1</name>

<value>weekend01:9000</value>

</property>

<!-- nn1的http通信地址 -->

<property>

<name>dfs.namenode.http-address.ns1.nn1</name>

<value>weekend01:50070</value>

</property>

<!-- nn2的RPC通信地址 -->

<property>

<name>dfs.namenode.rpc-address.ns1.nn2</name>

<value>weekend02:9000</value>

</property>

<!-- nn2的http通信地址 -->

<property>

<name>dfs.namenode.http-address.ns1.nn2</name>

<value>weekend02:50070</value>

</property>

<!-- 指定NameNode的元数据在JournalNode上的存放位置 -->

<property>

<name>dfs.namenode.shared.edits.dir</name>

<value>qjournal://weekend05:8485;weekend06:8485;weekend07:8485/ns1</value>

</property>

<!-- 指定JournalNode在本地磁盘存放数据的位置 -->

<property>

<name>dfs.journalnode.edits.dir</name>

<value>/home/hadoop/app/hadoop-2.4.1/journaldata</value>

</property>

<!-- 开启NameNode失败自动切换 -->

<property>

<name>dfs.ha.automatic-failover.enabled</name>

<value>true</value>

</property>

<!-- 配置失败自动切换实现方式 -->

<property>

<name>dfs.client.failover.proxy.provider.ns1</name>

<value>org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider</value>

</property>

<!-- 配置隔离机制方法，多个机制用换行分割，即每个机制暂用一行-->

<property>

<name>dfs.ha.fencing.methods</name>

<value>

sshfence

shell(/bin/true)

</value>

</property>

<!-- 使用sshfence隔离机制时需要ssh免登陆 -->

<property>

<name>dfs.ha.fencing.ssh.private-key-files</name>

<value>/home/hadoop/.ssh/id_rsa</value>

</property>

<!-- 配置sshfence隔离机制超时时间 -->

<property>

<name>dfs.ha.fencing.ssh.connect-timeout</name>

<value>30000</value>

</property>

/configuration>

export JAVA_HOME=/usr/java/jdk1.7.0_55

//告诉hbase使用外部的zk

export HBASE_MANAGES_ZK=false

<configuration>

<!-- 指定hbase在HDFS上存储的路径 -->

        <property>

                <name>hbase.rootdir</name>

                <value>hdfs://ns1/hbase</value>

        </property>

<!-- 指定hbase是分布式的 -->

        <property>

                <name>hbase.cluster.distributed</name>

                <value>true</value>

        </property>

<!-- 指定zk的地址，多个用“,”分割 -->

        <property>

                <name>hbase.zookeeper.quorum</name>

                <value>weekend05:2181,weekend06:2181,weekend07:2181</value>

        </property>

</configuration>

weekend03

weekend04

weekend05

weekend06

进入hbase命令行

./hbase shell

显示hbase中的表

list

创建user表，包含info、data两个列族

create 'user', 'info1', 'data1'

create 'user', {NAME => 'info', VERSIONS => '3'}

向user表中插入信息，row key为rk0001，列族info中添加name列标示符，值为zhangsan

put 'user', 'rk0001', 'info:name', 'zhangsan'

向user表中插入信息，row key为rk0001，列族info中添加gender列标示符，值为female

put 'user', 'rk0001', 'info:gender', 'female'

向user表中插入信息，row key为rk0001，列族info中添加age列标示符，值为20

put 'user', 'rk0001', 'info:age', 20

向user表中插入信息，row key为rk0001，列族data中添加pic列标示符，值为picture

put 'user', 'rk0001', 'data:pic', 'picture'

获取user表中row key为rk0001的所有信息

get 'user', 'rk0001'

获取user表中row key为rk0001，info列族的所有信息

get 'user', 'rk0001', 'info'

获取user表中row key为rk0001，info列族的name、age列标示符的信息

get 'user', 'rk0001', 'info:name', 'info:age'

获取user表中row key为rk0001，info、data列族的信息

get 'user', 'rk0001', 'info', 'data'

get 'user', 'rk0001', {COLUMN => ['info', 'data']}

get 'user', 'rk0001', {COLUMN => ['info:name', 'data:pic']}

获取user表中row key为rk0001，列族为info，版本号最新5个的信息

get 'user', 'rk0001', {COLUMN => 'info', VERSIONS => 2}

get 'user', 'rk0001', {COLUMN => 'info:name', VERSIONS => 5}

get 'user', 'rk0001', {COLUMN => 'info:name', VERSIONS => 5, TIMERANGE => [1392368783980, 1392380169184]}

获取user表中row key为rk0001，cell的值为zhangsan的信息

get 'people', 'rk0001', {FILTER => "ValueFilter(=, 'binary:图片')"}

获取user表中row key为rk0001，列标示符中含有a的信息

get 'people', 'rk0001', {FILTER => "(QualifierFilter(=,'substring:a'))"}

put 'user', 'rk0002', 'info:name', 'fanbingbing'

put 'user', 'rk0002', 'info:gender', 'female'

put 'user', 'rk0002', 'info:nationality', '中国'

get 'user', 'rk0002', {FILTER => "ValueFilter(=, 'binary:中国')"}

查询user表中的所有信息

scan 'user'

查询user表中列族为info的信息

scan 'user', {COLUMNS => 'info'}

scan 'user', {COLUMNS => 'info', RAW => true, VERSIONS => 5}

scan 'persion', {COLUMNS => 'info', RAW => true, VERSIONS => 3}

查询user表中列族为info和data的信息

scan 'user', {COLUMNS => ['info', 'data']}

scan 'user', {COLUMNS => ['info:name', 'data:pic']}

查询user表中列族为info、列标示符为name的信息

scan 'user', {COLUMNS => 'info:name'}

查询user表中列族为info、列标示符为name的信息,并且版本最新的5个

scan 'user', {COLUMNS => 'info:name', VERSIONS => 5}

查询user表中列族为info和data且列标示符中含有a字符的信息

scan 'user', {COLUMNS => ['info', 'data'], FILTER => "(QualifierFilter(=,'substring:a'))"}

查询user表中列族为info，rk范围是[rk0001, rk0003)的数据

scan 'people', {COLUMNS => 'info', STARTROW => 'rk0001', ENDROW => 'rk0003'}

查询user表中row key以rk字符开头的

scan 'user',{FILTER=>"PrefixFilter('rk')"}

查询user表中指定范围的数据

scan 'user', {TIMERANGE => [1392368783980, 1392380169184]}

删除数据

删除user表row key为rk0001，列标示符为info:name的数据

delete 'people', 'rk0001', 'info:name'

删除user表row key为rk0001，列标示符为info:name，timestamp为1392383705316的数据

delete 'user', 'rk0001', 'info:name', 1392383705316

清空user表中的数据

truncate 'people'

修改表结构

首先停用user表（新版本不用）

disable 'user'

添加两个列族f1和f2

alter 'people', NAME => 'f1'

alter 'user', NAME => 'f2'

启用表

enable 'user'

###disable 'user'(新版本不用)

删除一个列族：

alter 'user', NAME => 'f1', METHOD => 'delete' 或 alter 'user', 'delete' => 'f1'

添加列族f1同时删除列族f2

alter 'user', {NAME => 'f1'}, {NAME => 'f2', METHOD => 'delete'}

将user表的f1列族版本号改为5

alter 'people', NAME => 'info', VERSIONS => 5

启用表

enable 'user'

删除表

disable 'user'

drop 'user'

get 'person', 'rk0001', {FILTER => "ValueFilter(=, 'binary:中国')"}

get 'person', 'rk0001', {FILTER => "(QualifierFilter(=,'substring:a'))"}

scan 'person', {COLUMNS => 'info:name'}

scan 'person', {COLUMNS => ['info', 'data'], FILTER => "(QualifierFilter(=,'substring:a'))"}

scan 'person', {COLUMNS => 'info', STARTROW => 'rk0001', ENDROW => 'rk0003'}

scan 'person', {COLUMNS => 'info', STARTROW => '20140201', ENDROW => '20140301'}

scan 'person', {COLUMNS => 'info:name', TIMERANGE => [1395978233636, 1395987769587]}

delete 'person', 'rk0001', 'info:name'

alter 'person', NAME => 'ffff'

alter 'person', NAME => 'info', VERSIONS => 10

get 'user', 'rk0002', {COLUMN => ['info:name', 'data:pic']}

Algorithm

% remaining

Encoding

Decoding

GZIP

13.4%

21 MB/s

118 MB/s

LZO

20.5%

135 MB/s

410 MB/s

Zippy/Snappy

22.2%

172 MB/s

409 MB/s

public class HbaseDemo {

private Configuration conf = null;

@Before

public void init(){

conf = HBaseConfiguration.create();

conf.set("hbase.zookeeper.quorum", "weekend05,weekend06,weekend07");

}

@Test

public void testDrop() throws Exception{

HBaseAdmin admin = new HBaseAdmin(conf);

admin.disableTable("account");

admin.deleteTable("account");

admin.close();

}

@Test

public void testPut() throws Exception{

HTable table = new HTable(conf, "person_info");

Put p = new Put(Bytes.toBytes("person_rk_bj_zhang_000002"));

p.add("base_info".getBytes(), "name".getBytes(), "zhangwuji".getBytes());

table.put(p);

table.close();

}

@Test

public void testDel() throws Exception{

HTable table = new HTable(conf, "user");

Delete del = new Delete(Bytes.toBytes("rk0001"));

del.deleteColumn(Bytes.toBytes("data"), Bytes.toBytes("pic"));

table.delete(del);

table.close();

}

@Test

public void testGet() throws Exception{

HTable table = new HTable(conf, "person_info");

Get get = new Get(Bytes.toBytes("person_rk_bj_zhang_000001"));

get.setMaxVersions(5);

Result result = table.get(get);

List<Cell> cells = result.listCells();

for(Cell c:cells){

}

//result.getValue(family, qualifier);  可以从result中直接取出一个特定的value

//遍历出result中所有的键值对

List<KeyValue> kvs = result.list();

//kv  ---> f1:title:superise....      f1:author:zhangsan    f1:content:asdfasldgkjsldg

for(KeyValue kv : kvs){

String family = new String(kv.getFamily());

System.out.println(family);

String qualifier = new String(kv.getQualifier());

System.out.println(qualifier);

System.out.println(new String(kv.getValue()));

}

table.close();

}

/**

 * 多种过滤条件的使用方法

 * @throws Exception

 */

@Test

public void testScan() throws Exception{

HTable table = new HTable(conf, "person_info".getBytes());

Scan scan = new Scan(Bytes.toBytes("person_rk_bj_zhang_000001"), Bytes.toBytes("person_rk_bj_zhang_000002"));

//前缀过滤器----针对行键

Filter filter = new PrefixFilter(Bytes.toBytes("rk"));

//行过滤器  ---针对行键

ByteArrayComparable rowComparator = new BinaryComparator(Bytes.toBytes("person_rk_bj_zhang_000001"));

RowFilter rf = new RowFilter(CompareOp.LESS_OR_EQUAL, rowComparator);

/**

         * 假设rowkey格式为：创建日期_发布日期_ID_TITLE

         * 目标：查找  发布日期  为  2014-12-21  的数据

         * sc.textFile("path").flatMap(line=>line.split("\t")).map(x=>(x,1)).reduceByKey(_+_).map((_(2),_(1))).sortByKey().map((_(2),_(1))).saveAsTextFile("")

         *

         *

         */

        rf = new RowFilter(CompareOp.EQUAL , new SubstringComparator("_2014-12-21_"));

//单值过滤器 1 完整匹配字节数组

new SingleColumnValueFilter("base_info".getBytes(), "name".getBytes(), CompareOp.EQUAL, "zhangsan".getBytes());

//单值过滤器2 匹配正则表达式

ByteArrayComparable comparator = new RegexStringComparator("zhang.");

new SingleColumnValueFilter("info".getBytes(), "NAME".getBytes(), CompareOp.EQUAL, comparator);

//单值过滤器3 匹配是否包含子串,大小写不敏感

comparator = new SubstringComparator("wu");

new SingleColumnValueFilter("info".getBytes(), "NAME".getBytes(), CompareOp.EQUAL, comparator);

//键值对元数据过滤-----family过滤----字节数组完整匹配

        FamilyFilter ff = new FamilyFilter(

                CompareOp.EQUAL ,

                new BinaryComparator(Bytes.toBytes("base_info"))   //表中不存在inf列族，过滤结果为空

                );

        //键值对元数据过滤-----family过滤----字节数组前缀匹配

        ff = new FamilyFilter(

                CompareOp.EQUAL ,

                new BinaryPrefixComparator(Bytes.toBytes("inf"))   //表中存在以inf打头的列族info，过滤结果为该列族所有行

                );

       //键值对元数据过滤-----qualifier过滤----字节数组完整匹配

        filter = new QualifierFilter(

                CompareOp.EQUAL ,

                new BinaryComparator(Bytes.toBytes("na"))   //表中不存在na列，过滤结果为空

                );

        filter = new QualifierFilter(

                CompareOp.EQUAL ,

                new BinaryPrefixComparator(Bytes.toBytes("na"))   //表中存在以na打头的列name，过滤结果为所有行的该列数据

         );

        //基于列名(即Qualifier)前缀过滤数据的ColumnPrefixFilter

        filter = new ColumnPrefixFilter("na".getBytes());

        //基于列名(即Qualifier)多个前缀过滤数据的MultipleColumnPrefixFilter

        byte[][] prefixes = new byte[][] {Bytes.toBytes("na"), Bytes.toBytes("me")};

        filter = new MultipleColumnPrefixFilter(prefixes);

        //为查询设置过滤条件

        scan.setFilter(filter);

scan.addFamily(Bytes.toBytes("base_info"));

//一行

// Result result = table.get(get);

//多行的数据

ResultScanner scanner = table.getScanner(scan);

for(Result r : scanner){

/**

for(KeyValue kv : r.list()){

String family = new String(kv.getFamily());

System.out.println(family);

String qualifier = new String(kv.getQualifier());

System.out.println(qualifier);

System.out.println(new String(kv.getValue()));

}

*/

//直接从result中取到某个特定的value

byte[] value = r.getValue(Bytes.toBytes("base_info"), Bytes.toBytes("name"));

System.out.println(new String(value));

}

table.close();

}

/**

public abstract class TableMapper<KEYOUT, VALUEOUT>

extends Mapper<ImmutableBytesWritable, Result, KEYOUT, VALUEOUT> {

}

 * @author duanhaitao@itcast.cn

 *

 */

public class HbaseReader {

public static String flow_fields_import = "flow_fields_import";

static class HdfsSinkMapper extends TableMapper<Text, NullWritable>{

@Override

protected void map(ImmutableBytesWritable key, Result value, Context context) throws IOException, InterruptedException {

byte[] bytes = key.copyBytes();

String phone = new String(bytes);

byte[] urlbytes = value.getValue("f1".getBytes(), "url".getBytes());

String url = new String(urlbytes);

context.write(new Text(phone + "\t" + url), NullWritable.get());

}

}

static class HdfsSinkReducer extends Reducer<Text, NullWritable, Text, NullWritable>{

@Override

protected void reduce(Text key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {

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

}

}

public static void main(String[] args) throws Exception {

Configuration conf = HBaseConfiguration.create();

conf.set("hbase.zookeeper.quorum", "spark01");

Job job = Job.getInstance(conf);

job.setJarByClass(HbaseReader.class);

// job.setMapperClass(HdfsSinkMapper.class);

Scan scan = new Scan();

TableMapReduceUtil.initTableMapperJob(flow_fields_import, scan, HdfsSinkMapper.class, Text.class, NullWritable.class, job);

job.setReducerClass(HdfsSinkReducer.class);

FileOutputFormat.setOutputPath(job, new Path("c:/hbasetest/output"));

job.setOutputKeyClass(Text.class);

job.setOutputValueClass(NullWritable.class);

job.waitForCompletion(true);

}

}

/**

public abstract class TableReducer<KEYIN, VALUEIN, KEYOUT>

extends Reducer<KEYIN, VALUEIN, KEYOUT, Writable> {

}

 * @author duanhaitao@itcast.cn

 *

 */

public class HbaseSinker {

public static String flow_fields_import = "flow_fields_import";

static class HbaseSinkMrMapper extends Mapper<LongWritable, Text, FlowBean, NullWritable>{

@Override

protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

String line = value.toString();

String[] fields = line.split("\t");

String phone = fields[0];

String url = fields[1];

FlowBean bean = new FlowBean(phone,url);

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

}

}

static class HbaseSinkMrReducer extends TableReducer<FlowBean, NullWritable, ImmutableBytesWritable>{

@Override

protected void reduce(FlowBean key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {

Put put = new Put(key.getPhone().getBytes());

put.add("f1".getBytes(), "url".getBytes(), key.getUrl().getBytes());

context.write(new ImmutableBytesWritable(key.getPhone().getBytes()), put);

}

}

public static void main(String[] args) throws Exception {

Configuration conf = HBaseConfiguration.create();

conf.set("hbase.zookeeper.quorum", "spark01");

HBaseAdmin hBaseAdmin = new HBaseAdmin(conf);

boolean tableExists = hBaseAdmin.tableExists(flow_fields_import);

if(tableExists){

hBaseAdmin.disableTable(flow_fields_import);

hBaseAdmin.deleteTable(flow_fields_import);

}

HTableDescriptor desc = new HTableDescriptor(TableName.valueOf(flow_fields_import));

HColumnDescriptor hColumnDescriptor = new HColumnDescriptor ("f1".getBytes());

desc.addFamily(hColumnDescriptor);

hBaseAdmin.createTable(desc);

Job job = Job.getInstance(conf);

job.setJarByClass(HbaseSinker.class);

job.setMapperClass(HbaseSinkMrMapper.class);

TableMapReduceUtil.initTableReducerJob(flow_fields_import, HbaseSinkMrReducer.class, job);

FileInputFormat.setInputPaths(job, new Path("c:/hbasetest/data"));

job.setMapOutputKeyClass(FlowBean.class);

job.setMapOutputValueClass(NullWritable.class);

job.setOutputKeyClass(ImmutableBytesWritable.class);

job.setOutputValueClass(Mutation.class);

job.waitForCompletion(true);

}

}

<configuration>

<property>

<name>javax.jdo.option.ConnectionURL</name>

<value>jdbc:mysql://weekend01:3306/hive?createDatabaseIfNotExist=true</value>

</property>

<property>

<name>javax.jdo.option.ConnectionDriverName</name>

<value>com.mysql.jdbc.Driver</value>

</property>

<property>

<name>javax.jdo.option.ConnectionUserName</name>

<value>root</value>

</property>

<property>

<name>javax.jdo.option.ConnectionPassword</name>

<value>root</value>

</property>

</configuration>

set hive.cli.print.header=true;

CREATE TABLE page_view(viewTime INT, userid BIGINT,

     page_url STRING, referrer_url STRING,

     ip STRING COMMENT 'IP Address of the User')

 COMMENT 'This is the page view table'

 PARTITIONED BY(dt STRING, country STRING)

 ROW FORMAT DELIMITED

   FIELDS TERMINATED BY '\001'

STORED AS SEQUENCEFILE;

//sequencefile

create table tab_ip_seq(id int,name string,ip string,country string)

    row format delimited

    fields terminated by ','

    stored as sequencefile;

//使用select语句来批量插入数据

insert overwrite table tab_ip_seq select * from tab_ext;

//create & load

create table tab_ip(id int,name string,ip string,country string)

    row format delimited

    fields terminated by ','

    stored as textfile;

//从本地导入数据到hive的表中（实质就是将文件上传到hdfs中hive管理目录下）

load data local inpath '/home/hadoop/ip.txt' into table tab_ext;

//从hdfs上导入数据到hive表中（实质就是将文件从原始目录移动到hive管理的目录下）

load data inpath 'hdfs://ns1/aa/bb/data.log' into table tab_user;

//external外部表

CREATE EXTERNAL TABLE tab_ip_ext(id int, name string,

     ip STRING,

     country STRING)

 ROW FORMAT DELIMITED FIELDS TERMINATED BY ','

 STORED AS TEXTFILE

 LOCATION '/external/user';

// CTAS  根据select语句建表结构

CREATE TABLE tab_ip_ctas

   AS

SELECT id new_id, name new_name, ip new_ip,country new_country

FROM tab_ip_ext

SORT BY new_id;

//CLUSTER <--相对高级一点，你可以放在有精力的时候才去学习>

create table tab_ip_cluster(id int,name string,ip string,country string)

clustered by(id) into 3 buckets;

load data local inpath '/home/hadoop/ip.txt' overwrite into table tab_ip_cluster;

set hive.enforce.bucketing=true;

insert into table tab_ip_cluster select * from tab_ip;

select * from tab_ip_cluster tablesample(bucket 2 out of 3 on id);

//PARTITION  分区表

create table tab_ip_part(id int,name string,ip string,country string)

    partitioned by (year string)

    row format delimited fields terminated by ',';

load data local inpath '/home/hadoop/data.log' overwrite into table tab_ip_part

     partition(year='1990');

load data local inpath '/home/hadoop/data2.log' overwrite into table tab_ip_part

     partition(year='2000');

select * from tab_ip_part;

select * from tab_ip_part  where part_flag='part2';

select count(*) from tab_ip_part  where part_flag='part2';

alter table tab_ip change id id_alter string;

ALTER TABLE tab_cts ADD PARTITION (partCol = 'dt') location '/external/hive/dt';

show partitions tab_ip_part;

//insert from select   通过select语句批量插入数据到别的表

create table tab_ip_like like tab_ip;

insert overwrite table tab_ip_like

    select * from tab_ip;

//write to hdfs  将结果写入到hdfs的文件中

insert overwrite local directory '/home/hadoop/hivetemp/test.txt' select * from tab_ip_part where part_flag='part1';    

insert overwrite directory '/hiveout.txt' select * from tab_ip_part where part_flag='part1';

//cli shell  通过shell执行hive的hql语句

hive -S -e 'select country,count(*) from tab_ext' > /home/hadoop/hivetemp/e.txt  

select * from tab_ext sort by id desc limit 5;

select a.ip,b.book from tab_ext a join tab_ip_book b on(a.name=b.name);

//array

create table tab_array(a array<int>,b array<string>)

row format delimited

fields terminated by '\t'

collection items terminated by ',';

select a[0] from tab_array;

select * from tab_array where array_contains(b,'word');

insert into table tab_array select array(0),array(name,ip) from tab_ext t;

//map

create table tab_map(name string,info map<string,string>)

row format delimited

fields terminated by '\t'

collection items terminated by ','

map keys terminated by ':';

load data local inpath '/home/hadoop/hivetemp/tab_map.txt' overwrite into table tab_map;

insert into table tab_map select name,map('name',name,'ip',ip) from tab_ext;

//struct

create table tab_struct(name string,info struct<age:int,tel:string,addr:string>)

row format delimited

fields terminated by '\t'

collection items terminated by ','

load data local inpath '/home/hadoop/hivetemp/tab_st.txt' overwrite into table tab_struct;

insert into table tab_struct select name,named_struct('age',id,'tel',name,'addr',country) from tab_ext;

//UDF

select if(id=1,first,no-first),name from tab_ext;

hive>add jar /home/hadoop/myudf.jar;

hive>CREATE TEMPORARY FUNCTION fanyi AS 'cn.itcast.hive.Fanyi';

select id,name,ip,fanyi(country) from tab_ip_ext;

root@m1:/home/hadoop/flume-1.5.0-bin# cp conf/flume-env.sh.template conf/flume-env.sh

root@m1:/home/hadoop/flume-1.5.0-bin# vi conf/flume-env.sh

# Licensed to the Apache Software Foundation (ASF) under one

# or more contributor license agreements.  See the NOTICE file

# distributed with this work for additional information

# regarding copyright ownership.  The ASF licenses this file

# to you under the Apache License, Version 2.0 (the

# "License"); you may not use this file except in compliance

# with the License.  You may obtain a copy of the License at

#

#     http://www.apache.org/licenses/LICENSE-2.0

#

# Unless required by applicable law or agreed to in writing, software

# distributed under the License is distributed on an "AS IS" BASIS,

# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

# See the License for the specific language governing permissions and

# limitations under the License.

# If this file is placed at FLUME_CONF_DIR/flume-env.sh, it will be sourced

# during Flume startup.

# Enviroment variables can be set here.

JAVA_HOME=/usr/lib/jvm/java-7-oracle

# Give Flume more memory and pre-allocate, enable remote monitoring via JMX

#JAVA_OPTS="-Xms100m -Xmx200m -Dcom.sun.management.jmxremote"

# Note that the Flume conf directory is always included in the classpath.

#FLUME_CLASSPATH=""

root@m1:/home/hadoop# /home/hadoop/flume-1.5.0-bin/bin/flume-ng version

Flume 1.5.0

Source code repository: https://git-wip-us.apache.org/repos/asf/flume.git

Revision: 8633220df808c4cd0c13d1cf0320454a94f1ea97

Compiled by hshreedharan on Wed May  7 14:49:18 PDT 2014

From source with checksum a01fe726e4380ba0c9f7a7d222db961f

root@m1:/home/hadoop#

root@m1:/home/hadoop#vi /home/hadoop/flume-1.5.0-bin/conf/avro.conf

a1.sources = r1

a1.sinks = k1

a1.channels = c1

# Describe/configure the source

a1.sources.r1.type = avro

a1.sources.r1.channels = c1

a1.sources.r1.bind = 0.0.0.0

a1.sources.r1.port = 4141

# Describe the sink

a1.sinks.k1.type = logger

# Use a channel which buffers events in memory

a1.channels.c1.type = memory

a1.channels.c1.capacity = 1000

a1.channels.c1.transactionCapacity = 100

# Bind the source and sink to the channel

a1.sources.r1.channels = c1

a1.sinks.k1.channel = c1

root@m1:/home/hadoop# /home/hadoop/flume-1.5.0-bin/bin/flume-ng agent -c . -f /home/hadoop/flume-1.5.0-bin/conf/avro.conf -n a1 -Dflume.root.logger=INFO,console

root@m1:/home/hadoop# echo "hello world" > /home/hadoop/flume-1.5.0-bin/log.00

root@m1:/home/hadoop# /home/hadoop/flume-1.5.0-bin/bin/flume-ng avro-client -c . -H m1 -p 4141 -F /

root@m1:/home/hadoop/flume-1.5.0-bin/conf# /home/hadoop/flume-1.5.0-bin/bin/flume-ng agent -c . -f /home/hadoop/flume-1.5.0-bin/conf/avro.conf -n a1 -Dflume.root.logger=INFO,console

Info: Sourcing environment configuration script /home/hadoop/flume-1.5.0-bin/conf/flume-env.sh

Info: Including Hadoop libraries found via (/home/hadoop/hadoop-2.2.0/bin/hadoop) for HDFS access

Info: Excluding /home/hadoop/hadoop-2.2.0/share/hadoop/common/lib/slf4j-api-1.7.5.jar from classpath

Info: Excluding /home/hadoop/hadoop-2.2.0/share/hadoop/common/lib/slf4j-log4j12-1.7.5.jar from classpath

...

2014-08-10 10:43:25,112 (New I/O  worker #1) [INFO - org.apache.avro.ipc.NettyServer$NettyServerAvroHandler.handleUpstream(NettyServer.java:171)] [id: 0x92464c4f, /192.168.1.50:59850 :> /192.168.1.50:4141] UNBOUND

2014-08-10 10:43:25,112 (New I/O  worker #1) [INFO - org.apache.avro.ipc.NettyServer$NettyServerAvroHandler.handleUpstream(NettyServer.java:171)] [id: 0x92464c4f, /192.168.1.50:59850 :> /192.168.1.50:4141] CLOSED

2014-08-10 10:43:25,112 (New I/O  worker #1) [INFO - org.apache.avro.ipc.NettyServer$NettyServerAvroHandler.channelClosed(NettyServer.java:209)] Connection to /192.168.1.50:59850 disconnected.

2014-08-10 10:43:26,718 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:70)] Event: { headers:{} body: 68 65 6C 6C 6F 20 77 6F 72 6C 64                hello world }

root@m1:/home/hadoop# vi /home/hadoop/flume-1.5.0-bin/conf/spool.conf

a1.sources = r1

a1.sinks = k1

a1.channels = c1

# Describe/configure the source

a1.sources.r1.type = spooldir

a1.sources.r1.channels = c1

a1.sources.r1.spoolDir = /home/hadoop/flume-1.5.0-bin/logs

a1.sources.r1.fileHeader = true

# Describe the sink

a1.sinks.k1.type = logger

# Use a channel which buffers events in memory

a1.channels.c1.type = memory

a1.channels.c1.capacity = 1000

a1.channels.c1.transactionCapacity = 100

# Bind the source and sink to the channel

a1.sources.r1.channels = c1

a1.sinks.k1.channel = c1

root@m1:/home/hadoop# /home/hadoop/flume-1.5.0-bin/bin/flume-ng agent -c . -f /home/hadoop/flume-1.5.0-bin/conf/spool.conf -n a1 -Dflume.root.logger=INFO,console

root@m1:/home/hadoop# echo "spool test1" > /home/hadoop/flume-1.5.0-bin/logs/spool_text.log

14/08/10 11:37:13 INFO source.SpoolDirectorySource: Spooling Directory Source runner has shutdown.

14/08/10 11:37:13 INFO source.SpoolDirectorySource: Spooling Directory Source runner has shutdown.

14/08/10 11:37:14 INFO avro.ReliableSpoolingFileEventReader: Preparing to move file /home/hadoop/flume-1.5.0-bin/logs/spool_text.log to /home/hadoop/flume-1.5.0-bin/logs/spool_text.log.COMPLETED

14/08/10 11:37:14 INFO source.SpoolDirectorySource: Spooling Directory Source runner has shutdown.

14/08/10 11:37:14 INFO source.SpoolDirectorySource: Spooling Directory Source runner has shutdown.

14/08/10 11:37:14 INFO sink.LoggerSink: Event: { headers:{file=/home/hadoop/flume-1.5.0-bin/logs/spool_text.log} body: 73 70 6F 6F 6C 20 74 65 73 74 31                spool test1 }

14/08/10 11:37:15 INFO source.SpoolDirectorySource: Spooling Directory Source runner has shutdown.

14/08/10 11:37:15 INFO source.SpoolDirectorySource: Spooling Directory Source runner has shutdown.

14/08/10 11:37:16 INFO source.SpoolDirectorySource: Spooling Directory Source runner has shutdown.

14/08/10 11:37:16 INFO source.SpoolDirectorySource: Spooling Directory Source runner has shutdown.

14/08/10 11:37:17 INFO source.SpoolDirectorySource: Spooling Directory Source runner has shutdown.

root@m1:/home/hadoop# vi /home/hadoop/flume-1.5.0-bin/conf/exec_tail.conf

a1.sources = r1

a1.sinks = k1

a1.channels = c1

# Describe/configure the source

a1.sources.r1.type = exec

a1.sources.r1.channels = c1

a1.sources.r1.command = tail -F /home/hadoop/flume-1.5.0-bin/log_exec_tail

# Describe the sink

a1.sinks.k1.type = logger

# Use a channel which buffers events in memory

a1.channels.c1.type = memory

a1.channels.c1.capacity = 1000

a1.channels.c1.transactionCapacity = 100

# Bind the source and sink to the channel

a1.sources.r1.channels = c1

a1.sinks.k1.channel = c1

root@m1:/home/hadoop# /home/hadoop/flume-1.5.0-bin/bin/flume-ng agent -c . -f /home/hadoop/flume-1.5.0-bin/conf/exec_tail.conf -n a1 -Dflume.root.logger=INFO,console

root@m1:/home/hadoop# for i in {1..100};do echo "exec tail$i" >> /home/hadoop/flume-1.5.0-bin/log_

2014-08-10 10:59:25,513 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:70)] Event: { headers:{} body: 65 78 65 63 20 74 61 69 6C 20 74 65 73 74       exec tail test }

2014-08-10 10:59:34,535 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:70)] Event: { headers:{} body: 65 78 65 63 20 74 61 69 6C 20 74 65 73 74       exec tail test }

2014-08-10 11:01:40,557 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:70)] Event: { headers:{} body: 65 78 65 63 20 74 61 69 6C 31                   exec tail1 }

2014-08-10 11:01:41,180 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:70)] Event: { headers:{} body: 65 78 65 63 20 74 61 69 6C 32                   exec tail2 }

2014-08-10 11:01:41,180 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:70)] Event: { headers:{} body: 65 78 65 63 20 74 61 69 6C 33                   exec tail3 }

2014-08-10 11:01:41,181 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:70)] Event: { headers:{} body: 65 78 65 63 20 74 61 69 6C 34                   exec tail4 }

2014-08-10 11:01:41,181 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:70)] Event: { headers:{} body: 65 78 65 63 20 74 61 69 6C 35                   exec tail5 }

2014-08-10 11:01:41,181 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:70)] Event: { headers:{} body: 65 78 65 63 20 74 61 69 6C 36                   exec tail6 }

....

....

....

2014-08-10 11:01:51,550 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:70)] Event: { headers:{} body: 65 78 65 63 20 74 61 69 6C 39 36                exec tail96 }

2014-08-10 11:01:51,550 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:70)] Event: { headers:{} body: 65 78 65 63 20 74 61 69 6C 39 37                exec tail97 }

2014-08-10 11:01:51,551 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:70)] Event: { headers:{} body: 65 78 65 63 20 74 61 69 6C 39 38                exec tail98 }

2014-08-10 11:01:51,551 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:70)] Event: { headers:{} body: 65 78 65 63 20 74 61 69 6C 39 39                exec tail99 }

2014-08-10 11:01:51,551 (SinkRunner-PollingRunner-DefaultSinkProcessor) [INFO - org.apache.flume.sink.LoggerSink.process(LoggerSink.java:70)] Event: { headers:{} body: 65 78 65 63 20 74 61 69 6C 31 30 30             exec tail100 }

root@m1:/home/hadoop# vi /home/hadoop/flume-1.5.0-bin/conf/syslog_tcp.conf

a1.sources = r1

a1.sinks = k1

a1.channels = c1

# Describe/configure the source

a1.sources.r1.type = syslogtcp

a1.sources.r1.port = 5140