xml config lookup shell scripts

例如可以用下面的scripts來查dfs.replication的設定值為何 [root@master02.dev ~]>cat ~/tttt.sh #!/bin/bash SEARCHPATH="./" if [ -z "$1" ]; then SEARCHPATH="./" else SEARCHPATH="$1" fi for file in `find $SEARCHPATH -name "*xml"` do echo $file cat $file | awk -F "\n" 'BEGIN{a=-1}{if(a < 0) { match($1 , pattern , ar); if( RSTART > 0) a=range; } if( a > 0) { a--; print markhead,$1; } }' range=2 pattern='dfs.replication' markhead='#' done [root@master02.dev process]>cd [root@master02.dev ~]>cd /var/run/cloudera-scm-agent/process [root@master02.dev process]>~/tttt.sh ./12-mapreduce-TASKTRACKER/mapred-site.xml ./12-mapreduce-TASKTRACKER/core-site.xml ./12-mapreduce-TASKTRACKER/hdfs-site.xml # <name>dfs.replication</name> # <value>3</value> ./13-mapreduce-JOBTRACKER/mapred-queue-acls.xml ./13-mapreduce-JOBTRACKER/fair-scheduler.xml ./13-mapreduce-JOBTRACKER/mapred-site.xml ./13-mapreduce-JOBTRACKER/core-site.xml ./13-mapreduce-JOBTRACKER/capacity-scheduler.xml ./13-mapreduce-JOBTRACKER/hdfs-site.xml # <name>dfs.replication</name> # <value>3</value> ./9-hdfs-NAMENODE/core-site.xml ./9-hdfs-NAMENODE/hdfs-site.xml # <name>dfs.replication</name> # <value>3</value> ./19-deploy-client-config/hadoop-conf/core-site.xml ./19-deploy-client-config/hadoop-conf/hdfs-site.xml # <name>dfs.replication</name> # <value>3</value> ./10-hdfs-DATANODE/core-site.xml ./10-hdfs-DATANODE/hdfs-site.xml # <name>dfs.replication</name> # <value>3</value> ./7-hdfs-NAMENODE-format/core-site.xml ./7-hdfs-NAMENODE-format/hdfs-site.xml # <name>dfs.replication</name> # <value>3</value> ./20-deploy-client-config/hadoop-conf/mapred-site.xml ./20-deploy-client-config/hadoop-conf/core-site.xml ./20-deploy-client-config/hadoop-conf/hdfs-site.xml # <name>dfs.replication</name> # <value>3</value> ./11-hdfs-NAMENODE-createtmp/core-site.xml ./11-hdfs-NAMENODE-createtmp/hdfs-site.xml # <name>dfs.replication</name> # <value>3</value> ./8-hdfs-SECONDARYNAMENODE/core-site.xml ./8-hdfs-SECONDARYNAMENODE/hdfs-site.xml # <name>dfs.replication</name> # <value>3</value>

Easy Minutely Candle-Stick Data By MapReduce under CDH4.2.0 / CM4.5.0 --(3) Pipes

Environment:

OS: CentOS 6.2
Hadoop: CDH4.2.0 install via CM4.5.0
Hosts:
     Hostname  processes          ip
     ========  =================  ================
     master01  Namenode(CDH4)     192.168.122.11
     smaster   SecondaryNamenode  192.168.122.10
               JobTracker
     knode020  Datanode           192.168.122.20
               TaskTracker
     knode021  Datanode           192.168.122.21
               TaskTracker

MapReduce Version: 0.20.2


TWSE(Stock and warrant ticks)
  |
  +-> socket client(with libhdfs.so)
       |
       +-> HDFS(Stored by date)
             |
             +->MapReduce(candle-pipes.cc)
                  |
                  +-> HDFS(minutely candle stick data)

Map & Reduce Classes

candle-pipes.cc

#include <iostream>
#include <string>  // for std::stoi
#include <sstream> // for std::istringstream
#include <stdexcept> // for exception object
#include <vector>  // for vector<>


#include <hadoop/Pipes.hh>
#include <hadoop/TemplateFactory.hh>
#include <hadoop/StringUtils.hh>
const char kKVDelim = '\t';
const char kSecKDelim = ',';
const char kLineDelimitor = '\n';

const char* sKVDelim = "\t";
const char* sSecKDelim = ",";
const char* sLineDelimitor = "\n";

class CandlePipesMapper : public HadoopPipes::Mapper{
public:
  CandlePipesMapper(HadoopPipes::TaskContext& context){}
  void map(HadoopPipes::MapContext& context){
       std::string timestamp;
       std::string symbol;
       std::string SEQN;
       std::string HiLoMark;
       std::string MatchTime;
       std::string MatchPrice;
       std::string MatchVolumn;
       std::string line = context.getInputValue();

       std::vector< std::string > words = HadoopUtils::splitString( line, sKVDelim );
       if(words.size() > 6 && words[5] != "-") /*MatchPrice != "-"*/
       {
          timestamp = words[0];
          symbol = words[1];
          SEQN = words[2];
          MatchTime = words[3];
          HiLoMark = words[4];
          MatchPrice = words[5];
          MatchVolumn = words[6];

          context.emit(symbol + sKVDelim + MatchTime.substr(0,5),
             MatchTime.substr(6,2) + sKVDelim
                  + SEQN + sKVDelim
                  + timestamp + sKVDelim
                  + HiLoMark + sKVDelim
                  + MatchPrice + sKVDelim
                  + MatchVolumn);

       }

  }
};

class CandlePipesReducer : public HadoopPipes::Reducer{
public:
  CandlePipesReducer(HadoopPipes::ReduceContext& context){}
  void reduce(HadoopPipes::ReduceContext& context){
       std::string line;
       std::string token;
       int high = 0;
       int open = 0;
       int low = 0;
       int close = 0;
       int vol = 0;
       int open_seqn = 0;
       int close_seqn = 0;
       int seqn = 0;
       int b_received = 0;
       std::string last_symbol="";
       std::string last_MatchHHMM;
       while(context.nextValue())
       {
         int val=0;
         std::string values = context.getInputValue();

         std::vector< std::string > words = HadoopUtils::splitString( values, sKVDelim );
         if(words.size() >= 6 ) /*MatchPrice != "-"*/
         {
           try{
             /*match seqn*/
             seqn = std::stoi(words[1]);

             /*match price*/
             val = std::stoi(words[4]);
             if(open_seqn == 0 || open_seqn  > seqn)
             {
               open = val;
               open_seqn = seqn;
             }
             if(high == 0 || val > high)
               high = val;
             if(low == 0 || val < low)
               low = val;
             if(close_seqn == 0 || close_seqn < seqn)
             {
               close = val;
               close_seqn = seqn;
             }

             /*match volumn*/
             val = std::stoi(words[5]); /*total volumn*/
             if(val > vol)
               vol=val;
           }
           catch (const std::invalid_argument& ia) {
             std::cerr << "Invalid argument: " << ia.what() << '\n';
             continue;
           }
           catch (const std::out_of_range& oor) {
             std::cerr << "Out of Range error: " << oor.what() << '\n';
             continue;
           }


           b_received = 1;

         }


     }
     if(b_received)
     {
         context.emit(context.getInputKey(),
                  HadoopUtils::toString(open)  + sKVDelim
                  + HadoopUtils::toString(high) + sKVDelim
                  + HadoopUtils::toString(low) + sKVDelim
                  + HadoopUtils::toString(close) + sKVDelim
                  + HadoopUtils::toString(vol) );
     }

  }
};

int main(int argc , char *argv[])
{
  return HadoopPipes::runTask(HadoopPipes::TemplateFactory<CandlePipesMapper, CandlePipesReducer>());
}

Compile it:

[kccs@master01 candle-pipes]$ cat compile.sh
#!/bin/bash
make
if [ ! -e ~/bin/candle_stick ]; then
mkdir ~/bin/candle_stick
fi

#cp run2.sh ~/bin/candle_stick/.
##cp run-test.sh ~/bin/candle_stick/.
#cp envpre.sh ~/bin/candle_stick/.
#cp candle.jar ~/bin/candle_stick/.

[kccs@master01 candle-pipes]$ cat makefile
CC = g++
MAPREDUCE_INSTALL=/usr/lib/hadoop-0.20-mapreduce
CPPFLAGS = -m64 -I$(MAPREDUCE_INSTALL)/include

candle-pipes: candle-pipes.cc
        $(CC) $(CPPFLAGS) $< -Wall -L$(MAPREDUCE_INSTALL)/lib/native -std=c++0x -lhadooppipes -lhadooputils -lpthread -lssl -lcrypto -g -O2 -o $@
clean:
        rm -f *.o *.a *.s candle-pipes

## you need openssl-devel for libssl.so and libcrypto.so

Run it:

[kccs@master01 candle-pipes]$ cat run2.sh
#!/bin/bash
WHO=`whoami`
inputPath=/kccs/$1/
#inputPath=candle-input
outputPath=/user/$WHO/candle/$1
#outputPath=candle-output
program=candle-pipes

hdfs dfs -test -e $inputPath
if [ ! $? == "0" ]; then
echo "no tick data exist !!!"
exit 1
fi



hdfs dfs -test -e $outputPath
if [ $? == "0" ]; then
hdfs dfs -rm -r $outputPath
fi

hdfs dfs -test -d bin
if [ ! $? == "0" ]; then
hdfs dfs -mkdir bin
fi

hdfs dfs -rm ./candle-pipes bin/$program
hdfs dfs -put ./candle-pipes bin/$program

echo "candle stick start...."
hadoop pipes -D hadoop.pipes.java.recordreader=true -D hadoop.pipes.java.recordwriter=true -input $inputPath -output $outputPath -program bin/candle-pipes


[kccs@master01 candle-pipes]$ ./run2.sh YYYYMMDD
ex.
[kccs@master01 candle-pipes]$ ./run2.sh 20130801

Result:

[kccs@master01 candle]$ hdfs dfs -cat /user/kccs/candle/20130801/part-00000
:
:
:
9958    10:49   1265    1265    1265    1265    201
9958    10:51   1265    1265    1265    1265    247
9958    11:00   1260    1260    1260    1260    257
9958    11:03   1260    1260    1260    1260    313
9958    11:04   1260    1260    1260    1260    316
9958    11:10   1260    1260    1260    1260    318
9958    11:11   1260    1260    1260    1260    328
9958    11:14   1260    1260    1260    1260    336
9958    11:17   1260    1260    1260    1260    338
9958    11:19   1260    1260    1260    1260    340
9958    11:25   1260    1260    1260    1260    342
9958    11:30   1260    1260    1260    1260    358
9958    11:33   1255    1255    1255    1255    363
9958    11:35   1260    1260    1260    1260    364
9958    11:47   1255    1255    1255    1255    367
9958    11:54   1255    1255    1255    1255    368
9958    12:00   1260    1260    1260    1260    369
9958    12:06   1255    1255    1255    1255    394
9958    12:09   1255    1255    1255    1255    396
9958    12:25   1260    1260    1260    1260    397
9958    12:34   1255    1255    1255    1255    417
9958    12:35   1255    1255    1255    1255    418
9958    12:42   1255    1255    1255    1255    420
9958    12:50   1255    1255    1255    1255    430
9958    13:00   1255    1255    1255    1255    431
9958    13:01   1255    1255    1255    1255    432
9958    13:11   1255    1255    1255    1255    442
9958    13:12   1255    1255    1255    1255    452
9958    13:14   1255    1255    1255    1255    455
9958    13:17   1255    1255    1255    1255    460
9958    13:19   1255    1260    1255    1260    471
9958    13:24   1260    1260    1260    1260    480
9958    13:30   1260    1260    1260    1260    511

Compare:

run input-paths-to-process/trip-time(sec)/cpu-time(ms)
+--------------------------------------------------+
|run #    | Streaming   | Java       |    Pipes    |
|---------+-------------+------------+-------------+
| 1       | 5/28/-      | 5/29/14100 | 5/33/13180  |
| 2       | 5/26/-      | 5/24/16240 | 5/32/13540  |
| 3       | 5/25/-      | 5/27/16480 | 5/33/11540  |
| 4       | 5/26/-      | 5/29/13770 | 5/33/13460  |
| 5       | 5/31/-      | 5/24/16530 | 5/34/12440  |
|---------+-------------+------------+-------------+
|AVG      |   27.2      |   26.6     |   33.0      |
|run-trip |             |            |             |
+--------------------------------------------------+

memory physical/virtual
+----------------------------------------------------------------+
|run # | Streaming |         Java         |          Pipes       |
|------+-----------+----------------------+----------------------+
| 1    |    -      | 954920960/5369929728 | 863948800/5397696512 |
| 2    |    -      | 959614976/5370134528 | 860721152/5397913600 |
| 3    |    -      | 910581760/5689860096 | 878026752/5067251712 |
| 4    |    -      |1032314880/5074067456 | 854880256/5390655488 |
| 5    |    -      | 903376896/5691715584 | 848859136/5701148672 |
+----------------------------------------------------------------+

Easy Minutely Candle-Stick Data By MapReduce under CDH4.2.0 / CM4.5.0 --(2) Streaming

Environment:

OS: CentOS 6.2
Hadoop: CDH4.2.0 install via CM4.5.0
Hosts:
     Hostname  processes          ip
     ========  =================  ================
     master01  Namenode(CDH4)     192.168.122.11
     smaster   SecondaryNamenode  192.168.122.10
               JobTracker
     knode020  Datanode           192.168.122.20
               TaskTracker
     knode021  Datanode           192.168.122.21
               TaskTracker

MapReduce Version: 0.20.2


TWSE(Stock and warrant ticks)
  |
  +-> socket client(with libhdfs.so)
       |
       +-> HDFS(Stored by date)
             |
             +->MapReduce(candle-map, candle-reduce)
                  |
                  +-> HDFS(minutely candle stick data)

Map & Reduce Classes

candle-map.cc

#include <iostream>
#include <string>
#include <sstream>


const char kKVDelim = '\t';
const char kSecKDelim = ',';
const char kLineDelimitor = '\n';
class Mapper {
   public:
   Mapper(std::istream & input_stream){
       std::string timestamp;
       std::string symbol;
       std::string SEQN;
       std::string HiLoMark;
       std::string MatchTime;
       std::string MatchPrice;
       std::string MatchVolumn;
       std::string line;
       std::string token;
       while(getline(input_stream, line, kLineDelimitor))
       {
         int cnt = 0;
         std::istringstream issline(line);
         symbol=SEQN=MatchTime=HiLoMark=MatchPrice=MatchVolumn=timestamp="";
         while(getline(issline, token, kKVDelim) && cnt <= 6)
         {
           switch(cnt)
           {
           case 0:// timestamp
               timestamp = token;
               break;
           case 1:// symbol
               symbol = token;
               break;
           case 2:// SEQN
               SEQN = token;
               break;
           case 3:// MatchTime
               MatchTime = token;
               break;
           case 4:// HiLoMark
               HiLoMark = token;
               break;
           case 5:// MatchPrice
               MatchPrice = token;
               break;
           case 6:// MatchVolumn
               MatchVolumn = token;
               break;
           default: // order book
               break;
           }
           cnt++;
         }
         if(MatchPrice != "-")/*only match record*/
         {
           std::cout << symbol << kSecKDelim
              << MatchTime.substr(0,5) << kKVDelim
              << MatchTime.substr(6,2) << kKVDelim
              << SEQN << kKVDelim
              << timestamp << kKVDelim
              << HiLoMark << kKVDelim
              << MatchPrice << kKVDelim
              << MatchVolumn << std::endl;
         }
       }
   }
};
int main(int argc, char** argv) {
  Mapper mapper(std::cin);
  return 0;
}

candle-reduce.cc

#include <iostream>
#include <stdexcept>
#include <string>
#include <sstream>


const char kSecKDelim = ',';
const char kKVDelim = '\t';
const char kLineDelimitor = '\n';

class Reducer {
   public :
   Reducer(std::istream & input_stream){

       std::string line;
       std::string token;
       int high = 0;
       int open = 0;
       int low = 0;
       int close = 0;
       int vol = 0;
       int open_seqn = 0;
       int close_seqn = 0;
       int seqn = 0;
       int last_error = 0;
       std::string last_symbol="";
       std::string last_MatchHHMM;
       while(getline(input_stream, line, kLineDelimitor))
       {
         int cnt = 0;
         std::string symbol;
         std::string MatchHHMM;
         int val=0;
         last_error = 0;
         std::istringstream issline(line);
         while(getline(issline, token, kKVDelim) && cnt < 8)
         {
           try {
             switch(cnt)
             {
             case 0: //symbol,MatchHHMM
             {
               size_t delim_pos = token.find(kSecKDelim);
               if (delim_pos != std::string::npos) {
                 symbol = token.substr(0, delim_pos);
                 MatchHHMM = token.substr(delim_pos + 1);
               }
             }

               break;
             case 2: //SEQN
               seqn = std::stoi(token); // you need compile stoi with '-std=c++0x' option
               break;
             case 5:// MatchPrice
               val = std::stoi(token);
               if(open_seqn == 0 || open_seqn  > seqn)
               {
                 open = val;
                 open_seqn = seqn;
               }
               if(high == 0 || val > high)
                 high = val;
               if(low == 0 || val < low)
                 low = val;
               if(close_seqn == 0 || close_seqn < seqn)
               {
                 close = val;
                 close_seqn = seqn;
               }
               break;
             case 6:// MatchValumn
               val = std::stoi(token); /*total volumn*/
               if(val > vol)
                 vol=val;
               break;
             default:
               break;
             }
           }
           catch (const std::invalid_argument& ia) {
             std::cerr << "Invalid argument: " << ia.what() << '\n';
             last_error = 1;
             break;
           }
           catch (const std::out_of_range& oor) {
             std::cerr << "Out of Range error: " << oor.what() << '\n';
             last_error = 1;
             break;
           }

           if(  cnt==0 && (last_symbol != symbol || last_MatchHHMM != MatchHHMM))
           {
             if(last_symbol != "")
             {
               std::cout << last_symbol << kKVDelim
                  << last_MatchHHMM << kKVDelim
                  << open << kKVDelim
                  << high << kKVDelim
                  << low << kKVDelim
                  << close << kKVDelim
                  << vol << std::endl;
             }
             last_MatchHHMM = MatchHHMM;
             last_symbol = symbol;
             high = 0;
             open = 0;
             low = 0;
             close = 0;
             vol = 0;
             open_seqn = 0;
             close_seqn = 0;
             seqn = 0;
           }
           cnt++;
         }

       }
       if(last_error ==0)
       {

             std::cout << last_symbol << kKVDelim
                << last_MatchHHMM << kKVDelim
                << open << kKVDelim
                << high << kKVDelim
                << low << kKVDelim
                << close << kKVDelim
                << vol << std::endl;
       }
     }
   };
int main(int argc, char** argv) {
  Reducer reducer(std::cin);
  return 0;
}

Compile it:

[kccs@master01 candle-streaming]$ cat compile.sh
#!/bin/bash
g++ -o candle-map candle-map.cc
g++ -std=c++0x -o candle-reduce candle-reduce.cc

Run it:

[kccs@master01 candle-streaming]$ cat run2.sh
#!/bin/bash
WHO=`whoami`
inputPath=/kccs/$1/
outputPath=/user/$WHO/candle/$1
hdfs dfs -test -e $inputPath
if [ ! $? == "0" ]; then
echo "no tick data exist !!!"
exit 1
fi

WHO=`whoami`

hdfs dfs -test -e $outputPath
if [ $? == "0" ]; then
hdfs dfs -rm -r $outputPath
fi
echo "candle stick start...."
## you could test(debug) by local command
## cat [data source file] | ./candle-map | ./candle-reduce
hadoop jar /usr/lib/hadoop-0.20-mapreduce/contrib/streaming/hadoop-streaming-*.jar -file ./candle-map -mapper "./candle-map" -file ./candle-reduce -reducer ./candle-reduce -input $inputPath -output $outputPath

Result:

[kccs@master01 candle]$ hdfs dfs -cat /user/kccs/candle/20130730/part-00000
:
:
:
9958    12:24   1275    1280    1275    1280    543
9958    12:26   1280    1285    1280    1285    570
9958    12:27   1285    1285    1285    1285    575
9958    12:28   1285    1285    1285    1285    577
9958    12:30   1285    1285    1280    1280    580
9958    12:31   1280    1280    1280    1280    582
9958    12:32   1280    1280    1280    1280    587
9958    12:33   1280    1280    1280    1280    592
9958    12:34   1280    1280    1280    1280    612
9958    12:38   1275    1275    1275    1275    614
9958    12:43   1275    1275    1275    1275    621
9958    12:44   1275    1275    1275    1275    631
9958    12:45   1280    1280    1280    1280    632
9958    12:47   1275    1275    1275    1275    634
9958    12:51   1275    1275    1275    1275    644
9958    12:53   1275    1275    1275    1275    645
9958    12:58   1275    1275    1275    1275    647
9958    13:00   1275    1275    1275    1275    649
9958    13:01   1275    1275    1275    1275    652
9958    13:06   1280    1280    1280    1280    653
9958    13:10   1280    1280    1280    1280    673
9958    13:11   1280    1280    1280    1280    675
9958    13:12   1280    1280    1280    1280    680
9958    13:18   1275    1275    1275    1275    710
9958    13:19   1275    1275    1275    1275    717
9958    13:22   1275    1275    1275    1275    718
9958    13:23   1280    1280    1280    1280    723
9958    13:24   1280    1280    1280    1280    724
9958    13:30   1275    1275    1275    1275    737

Compare:

run trip time(sec)

+-----------------------------+

|run # | Streaming | Java     |

|------+-----------+----------|

| 1    |  28       |  24      |

| 2    |  24       |  24      |

| 3    |  23       |  22      |

| 4    |  25       |  26      |

| 5    |  31       |  25      |

|------+-----------+----------|

|AVG   |  26.2     |  24.2    | 

+-----------------------------+

Easy Minutely Candle-Stick Data By MapReduce under CDH4.2.0 / CM4.5.0--(1) JAVA

Environment:

OS: CentOS 6.2
Hadoop: CDH4.2.0 install via CM4.5.0
Hosts:
     Hostname  processes          ip
     ========  =================  ================
     master01  Namenode(CDH4)     192.168.122.11
     smaster   SecondaryNamenode  192.168.122.10
               JobTracker
     knode020  Datanode           192.168.122.20
               TaskTracker
     knode021  Datanode           192.168.122.21
               TaskTracker

MapReduce Version: 0.20.2

TWSE(Stock and warrant ticks)

  |

  +-> socket client(with libhdfs.so)

       |

       +-> HDFS(Stored by date)

             |

             +->MapReduce(Candle.java, candle.jar)

                  |

                  +-> HDFS(minutely candle stick data)

Map & Reduce Classes

   public static class Map extends MapReduceBase implements Mapper<LongWritable, Text, Text, Text> {

     private Text keytext = new Text();

     private Text valuetext = new Text();

     public void map(LongWritable key, Text value, OutputCollector<Text, Text> output, Reporter reporter) throws IOException {

       int cnt = 0;

       String line = value.toString();

       String timestamp = new String();

       String symbol = new String();

       String SEQN = new String();

       String HiLoMark = new String();

       String MatchTime = new String();

       String MatchPrice = new String();

       String MatchVolumn = new String();

       StringTokenizer tokenizer = new StringTokenizer(line,"\t\n"); // default delim = " \t\n\r\f"

       while (tokenizer.hasMoreTokens() && cnt <7 ) {

         switch(cnt)

         {

         case 0:// timestamp

             timestamp = tokenizer.nextToken();

             break;

         case 1:// symbol

             symbol = tokenizer.nextToken();

             break;

         case 2:// SEQN

             SEQN = tokenizer.nextToken();

             break;

         case 3:// MatchTime

             MatchTime = tokenizer.nextToken();

             break;

         case 4:// HiLoMark

             HiLoMark = tokenizer.nextToken();

             break;

         case 5:// MatchPrice

             MatchPrice = tokenizer.nextToken();

             break;

         case 6:// MatchVolumn

             break;

         case 6:// MatchVolumn

             MatchVolumn = tokenizer.nextToken();

             break;

         default: // order book

             tokenizer.nextToken();

             break;

         }

         cnt++;

       }

       if(!MatchPrice.matches("-"))/*only match record*/

       {

         //keytext.set( String.format("%s\t%s", symbol,MatchTime.substring(0,5)));

         keytext.set( String.format("%s\t%s", symbol,MatchTime.substring(0,5)));

         valuetext.set( String.format("%s\t%s\t%s\t%s\t%s\t%s",MatchTime.substring(6,8),SEQN,timestamp,HiLoMark,MatchPrice,MatchVolumn));

         output.collect(keytext, valuetext);

       }

     }

   }

   public static class Reduce extends MapReduceBase implements Reducer<Text, Text, Text, Text> {

     private int high = 0;

     private int open = 0;

     private int low = 0;

     private int close = 0;

     private int vol = 0;

     private int open_seqn = 0;

     private int close_seqn = 0;

     private Text candletext = new Text();

     public void reduce(Text key, Iterator<Text> values, OutputCollector<Text, Text> output, Reporter reporter) throws IOException {

       high = 0;

       open = 0;

       low = 0;

       close = 0;

       vol = 0;

       open_seqn = 0;

       close_seqn=0;

       while (values.hasNext()) {

         String Value=values.next().toString();

         StringTokenizer ValueTokenizer = new StringTokenizer(Value,"\t"); // default delim = " \t\n\r\f"

         int ii = 0;

         int seqn = 0;

         while (ValueTokenizer.hasMoreTokens() && ii <6 ) {

           int val = 0;

           String column = ValueTokenizer.nextToken();

           try

           {

             switch(ii++)

             {

             case 1: //SEQN

               seqn = Integer.parseInt(column);

               break;

             case 4:// MatchPrice

               val = Integer.parseInt(column);

               if(open_seqn == 0 || open_seqn  > seqn)

               {

                 open = val;

                 open_seqn = seqn;

               }

               if(high == 0 || val > high)

                 high = val;

               if(low == 0 || val < low)

                 low = val;

               if(close_seqn == 0 || close_seqn < seqn)

               {

                 close = val;

                 close_seqn = seqn;

               }

               break;

             case 5:// MatchValumn

               val = Integer.parseInt(column); /*total volumn*/

               if(val > vol)

                 vol=val;

               break;

             default:

               break;

             }

           }

           catch (NumberFormatException e)

           {

             System.err.println(e.toString());

             System.err.println("Integer.parseInt Error");

           }

           catch (Exception e)

           {

             System.err.println("Other Error"+e.getMessage());

           }

         }

       }

       candletext.set( String.format("%d\t%d\t%d\t%d\t%d",open,high,low,close,vol));

       output.collect(key, candletext);

     }

   }

                                                                                                       

Compile it:

[kccs@master01 candle]$ cat compile.sh

#!/bin/bash

if [ -e /usr/lib/bigtop-utils/bigtop-detect-javahome ]; then

        source /usr/lib/bigtop-utils/bigtop-detect-javahome

else

        source /usr/libexec/bigtop-detect-javahome

fi

export CLASSPATH=/etc/hadoop/conf

for file in `ls /usr/lib/hadoop/client/*.jar`

do

  export CLASSPATH=$CLASSPATH:$file

done

export CLASSPATH=$CLASSPATH:/usr/lib/hadoop/hadoop-annotations.jar

export LD_LIBRARY_PATH="$JAVA_HOME/jre/lib/amd64/server/"

if [ ! -e MyJava ]; then

mkdir MyJava

fi

javac -d MyJava Candle.java

jar -cvf candle.jar -C MyJava .

if [ ! -e ~/bin/candle_stick ]; then

mkdir ~/bin/candle_stick

fi

Run it:

[kccs@master01 candle]$ cat run2.sh

#!/bin/bash

WHO=`whoami`

inputPath=/kccs/$1/

outputPath=/user/$WHO/candle/$1

hdfs dfs -test -e $inputPath

if [ ! $? == "0" ]; then

echo "no tick data exist !!!"

exit 1

fi

WHO=`whoami`

hdfs dfs -test -e $outputPath

if [ $? == "0" ]; then

hdfs dfs -rm -r $outputPath

fi

echo "candle stick start...."

hadoop jar /home/kccs/src/ex_mapreduce/candle/candle.jar Candle $inputPath $outputPath

Result:

[kccs@master01 candle]$ hdfs dfs -cat /user/kccs/candle/20130730/part-00000

:

:

:

9958    12:24   1275    1280    1275    1280    543

9958    12:26   1280    1285    1280    1285    570

9958    12:27   1285    1285    1285    1285    575

9958    12:28   1285    1285    1285    1285    577

9958    12:30   1285    1285    1280    1280    580

9958    12:31   1280    1280    1280    1280    582

9958    12:32   1280    1280    1280    1280    587

9958    12:33   1280    1280    1280    1280    592

9958    12:34   1280    1280    1280    1280    612

9958    12:38   1275    1275    1275    1275    614

9958    12:43   1275    1275    1275    1275    621

9958    12:44   1275    1275    1275    1275    631

9958    12:45   1280    1280    1280    1280    632

9958    12:47   1275    1275    1275    1275    634

9958    12:51   1275    1275    1275    1275    644

9958    12:53   1275    1275    1275    1275    645

9958    12:58   1275    1275    1275    1275    647

9958    13:00   1275    1275    1275    1275    649

9958    13:01   1275    1275    1275    1275    652

9958    13:06   1280    1280    1280    1280    653

9958    13:10   1280    1280    1280    1280    673

9958    13:11   1280    1280    1280    1280    675

9958    13:12   1280    1280    1280    1280    680

9958    13:18   1275    1275    1275    1275    710

9958    13:19   1275    1275    1275    1275    717

9958    13:22   1275    1275    1275    1275    718

9958    13:23   1280    1280    1280    1280    723

9958    13:24   1280    1280    1280    1280    724

9958    13:30   1275    1275    1275    1275    737