这个是最开始的数据：乱七八糟的，要取出其中的一些，类似这些
其中毫秒级的时间数据要转为时间戳
spark先过滤出要取的数据
package sparkj; import java.text.SimpleDateFormat; import java.util.Arrays; import java.util.regex.Matcher; import java.util.regex.Pattern; import org.apache.spark.SparkConf; import org.apache.spark.api.java.JavaPairRDD; import org.apache.spark.api.java.JavaRDD; import org.apache.spark.api.java.JavaSparkContext; import scala.Tuple2; public class sp { public static void main(String[] args) { // TODO Auto-generated method stub SparkConf sparkConf = new SparkConf().setAppName("PeopleInfoCalculator").setMaster("local[2]"); JavaSparkContext sc = new JavaSparkContext(sparkConf); JavaRDD<String> filedata=sc.textFile("file:///root/555.txt"); JavaRDD<String> data2=filedata.filter(f->f.matches("[0-9].*")) .map(f->String2Date(f.split(" ")[0],f.split(" ")[1])+" "+f.split(" ")[2]+" "+f.split(" ")[3].split(":")[0]); data2.foreach(f->System.err.println(f)); //data2.saveAsTextFile("file:///root/555copy.txt"); // JavaPairRDD<String,Integer> rdd6=filedata.mapToPair(f->new Tuple2<>(f.split(",")[1],1)); // JavaPairRDD<String,Integer> rdd7=rdd6.reduceByKey((x,y)->x+y); //rdd7.foreach(f->System.err.println(f)); } public static long String2Date(String date,String time) throws Exception{ String newdate = date + " " + time; SimpleDateFormat s = new SimpleDateFormat("yyyy-MM-dd hh:mm:ss,SSS"); return s.parse(newdate).getTime(); } }
结果如下：
打开hbase,对hbase进行操作了，先start...,
root@master:/opt/hbase-1.3.3/bin# ./start-hbase.sh starting master, logging to /opt/hbase-1.3.3/bin/../logs/hbase-root-master-master.out Java HotSpot(TM) 64-Bit Server VM warning: ignoring option PermSize=128m; support was removed in 8.0 Java HotSpot(TM) 64-Bit Server VM warning: ignoring option MaxPermSize=128m; support was removed in 8.0 root@master:/opt/hbase-1.3.3/bin# ./hbase shell 2022-05-05 07:21:22,172 WARN [main] util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable HBase Shell; enter 'help<RETURN>' for list of supported commands. Type "exit<RETURN>" to leave the HBase Shell Version 1.3.3, ra75a458e0c8f3c60db57f30ec4b06a606c9027b4, Fri Dec 14 16:02:53 PST 2018 hbase(main):001:0> list TABLE
hbase单词计数
单独的hbase写法
主函数：
package test1; import java.io.BufferedReader; import java.io.File; import java.io.FileReader; import java.util.ArrayList; import java.util.Arrays; import java.util.List; import java.util.Scanner; public class MainHB1 { private static String getType(Object a) { return a.getClass().toString(); } public static void main(String[] args) throws Exception { // try { // HBaseOp ho = new HBaseOp(); // String tab = "logData1"; // String[] name = new String[1]; // name[0] = "log"; // ho.createTable(tab, name); // System.out.println("ok"); // } catch (Exception e) { // e.printStackTrace(); // } try { HBaseOp ho = new HBaseOp(); String tab = "logData1"; String[] name = new String[1]; name[0] = "log"; //ho.filterByRowFilter(tab, "class","info"); //ho.filterBySingleColumnValueFilter(tab, "log","level",""); System.out.println("ERROR一共是"); ho.filterBySingleColumnValueFilter(tab, "log","level","ERROR"); System.out.println("FATAL一共是"); ho.filterBySingleColumnValueFilter(tab, "log","level","FATAL"); System.out.println("ok"); } catch (Exception e) { e.printStackTrace(); } File file = new File("/root/555copy.txt"); System.out.println("插入数据成功"); System.out.println(" \\n" + "\\\n" + ""); BufferedReader br = new BufferedReader(new FileReader(file)); String s; String[] column = new String[2]; column[0] = "level"; column[1] = "class"; int i = 0; while ((s = br.readLine()) != null) { String line[] = s.split(" "); long t = Long.parseLong(line[0]); String time = String.valueOf(Long.MAX_VALUE-t); try { HBaseOp ho = new HBaseOp(); String tab = "logData1"; String name = "log"; // ho.put(tab,time, name,column[0], line[1]); // ho.put(tab,time, name,column[1], line[2]); } catch (Exception e) { e.printStackTrace(); } } System.out.println(i); System.out.println("ok"); } }
类的方法
package test1; import java.util.ArrayList; import java.util.HashMap; import java.util.List; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.HColumnDescriptor; import org.apache.hadoop.hbase.HTableDescriptor; import org.apache.hadoop.hbase.client.Delete; import org.apache.hadoop.hbase.client.Get; import org.apache.hadoop.hbase.client.HBaseAdmin; import org.apache.hadoop.hbase.client.HTable; import org.apache.hadoop.hbase.client.Put; import org.apache.hadoop.hbase.client.Result; import org.apache.hadoop.hbase.client.ResultScanner; import org.apache.hadoop.hbase.client.Scan; import org.apache.hadoop.hbase.filter.BinaryComparator; import org.apache.hadoop.hbase.filter.CompareFilter; //import org.apache.hadoop.hbase.filter.CompareOp; import org.apache.hadoop.hbase.filter.FilterList; import org.apache.hadoop.hbase.filter.SingleColumnValueFilter; import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp; import org.apache.hadoop.hbase.util.Bytes; public class HBaseOp { class P{ String row; String[] colum; } // TODO Auto-generated method stub Configuration conf=HBaseConfiguration.create(); public void createTable(String strTab,String[] arrCf) throws Exception{ HBaseAdmin admin=new HBaseAdmin(conf); if(admin.tableExists(strTab)) { System.out.println("Table"+strTab+"exists."); } else { HTableDescriptor fs=new HTableDescriptor(strTab); for(String cf:arrCf) { HColumnDescriptor ff=new HColumnDescriptor(cf); ff.setTimeToLive(10000); fs.addFamily(ff); } admin.createTable(fs); } admin.close(); } public void deleteTable(String strTab)throws Exception { HBaseAdmin admin = new HBaseAdmin(conf); if (!admin.tableExists(strTab)) {//判断表是否存在 System.out.println(strTab + "不存在"); } else if(admin.isTableEnabled(strTab)) {//如果表处于disable admin.disableTable(strTab); admin.deleteTable(strTab); System.out.println(strTab + " deleted"); } else { admin.deleteTable(strTab); System.out.println(strTab + " deleted"); } admin.close(); } // public void mulPut() throws Exception { HTable table = new HTable(conf,"scores"); // 创建一个列表用于存放Put实例 List<Put> puts = new ArrayList<Put>(); // 将第一个Put实例添加到列表 Put put1 = new Put(Bytes.toBytes("Tom")); put1.add(Bytes.toBytes("grade"), Bytes.toBytes(""), Bytes.toBytes("1")); put1.add(Bytes.toBytes("course"), Bytes.toBytes("math"), Bytes.toBytes("990")); puts.add(put1); // 将第2个Put实例添加到列表 Put put2 = new Put(Bytes.toBytes("John")); put2.add(Bytes.toBytes("grade"), Bytes.toBytes(""), Bytes.toBytes("2")); put2.add(Bytes.toBytes("course"), Bytes.toBytes("Chinese"), Bytes.toBytes("99")); puts.add(put2); // 将第3个Put实例添加到列表 Put put3 = new Put(Bytes.toBytes("gyy")); put3.add(Bytes.toBytes("grade"), Bytes.toBytes(""), Bytes.toBytes("12")); put3.add(Bytes.toBytes("course"), Bytes.toBytes("math"), Bytes.toBytes("1000")); put3.add(Bytes.toBytes("course"), Bytes.toBytes("geo"), Bytes.toBytes("1000")); put3.add(Bytes.toBytes("course"), Bytes.toBytes("语文"), Bytes.toBytes("1000")); // 向HBase中存入多行多列数据 puts.add(put3); table.put(puts); table.close(); } // public void put(String tablename,String row,String cf,String column,String data) throws Exception{ HTable table=new HTable(conf,tablename); Put p=new Put(Bytes.toBytes(row)); p.add(Bytes.toBytes(cf), Bytes.toBytes(column), Bytes.toBytes(data)); table.put(p); table.close(); } public void filterBySingleColumnValueFilter(String tablename, String strF, String strC, String strClass) throws Exception { HTable table = new HTable(conf, tablename); Scan s = new Scan(); SingleColumnValueFilter sf = new SingleColumnValueFilter(Bytes.toBytes(strF), Bytes.toBytes(strC), CompareOp.EQUAL, Bytes.toBytes(strClass)); s.setFilter(sf); ResultScanner rs = table.getScanner(s); HashMap<String, Integer> hashMap = new HashMap<>(); int i=0; for (Result r : rs) { i=i+1; byte[] row = r.getRow(); byte[] value = r.getValue(Bytes.toBytes(strF), Bytes.toBytes(strC)); byte[] value2 = r.getValue(Bytes.toBytes(strF), Bytes.toBytes("class")); if(!hashMap.containsKey(Bytes.toString(value2))) { //集合中没有该单词,值定义为1 hashMap.put(Bytes.toString(value2), 1); }else if(hashMap.containsKey(Bytes.toString(value2))) { //集合中有该单词,值+1;key不变; int b=hashMap.get(Bytes.toString(value2)); hashMap.put(Bytes.toString(value2), b+1); } // System.out.println("Filter: " + Bytes.toString(row) + " is in " + strC + " " + Bytes.toString(value)+", class " + Bytes.toString(value2)); } System.out.println(hashMap); // System.out.println( " ERROR一共是="+i); System.out.println("出现最多的类是："+getProcessCdByName(hashMap)); System.out.println(" FATAL一共是="+i); rs.close(); table.close(); } public static String getProcessCdByName(HashMap<String, Integer> processMap){//我找最大值对应的哪一个键 int max=0; for (Integer in : processMap.values()) { System.err.println(in); max=Math.max(max, in); } String result = null; Set<Map.Entry<String, Integer>> set = processMap.entrySet(); for(Map.Entry<String, Integer> entry : set){ if(entry.getValue()==max){ result = entry.getKey(); break; } } return result; } public void filterBySingleColumnValueFilter(String tablename, String cf, String C) throws Exception { HTable table = new HTable(conf, tablename); Scan s = new Scan(); SingleColumnValueFilter sf = new SingleColumnValueFilter(Bytes.toBytes(cf), Bytes.toBytes(C), CompareOp.EQUAL, new BinaryComparator(Bytes.toBytes("ERROR"))); SingleColumnValueFilter sf2 = new SingleColumnValueFilter(Bytes.toBytes(cf), Bytes.toBytes(C), CompareOp.EQUAL, new BinaryComparator(Bytes.toBytes("FATAL"))); FilterList lst= new FilterList( FilterList.Operator.MUST_PASS_ONE); lst.addFilter(sf); lst.addFilter(sf2); s.setFilter(lst); ResultScanner rs = table.getScanner(s); int i =0; for (Result r : rs) { byte[] row = r.getRow(); byte[] value = r.getValue(Bytes.toBytes(cf), Bytes.toBytes(C)); byte[] value2 = r.getValue(Bytes.toBytes(cf), Bytes.toBytes("class")); System.out.println("--------------------------"); System.out.println("Filter: " + Bytes.toString(row) + " is in " + C + " " + Bytes.toString(value)+", class " + Bytes.toString(value2)); i=i+1; } System.out.println(i); rs.close(); table.close(); } //public void filterBySingleColumnValueFilter(String tablename, String strF, String strC, String strClass) throws Exception //{ // HTable table = new HTable(conf, tablename); // Scan s = new Scan(); // SingleColumnValueFilter sf = // new SingleColumnValueFilter(Bytes.toBytes(strF), // Bytes.toBytes(strC), CompareOp.EQUAL, // Bytes.toBytes(strClass)); // s.setFilter(sf); // // ResultScanner rs = table.getScanner(s); // // for (Result r : rs) { // byte[] row = r.getRow(); // byte[] value = r.getValue(Bytes.toBytes(strF), Bytes.toBytes(strC)); // byte[] value2 = r.getValue(Bytes.toBytes(strF), Bytes.toBytes("class")); // System.out.println("Filter: " + Bytes.toString(row) + " is in " + strC + " " + Bytes.toString(value)+", class " + Bytes.toString(value2)); // } // rs.close(); // table.close(); //} public void filterBySingleColumnValueFilter(String tablename, String strF, String strC, String strClass) throws Exception { HTable table = new HTable(conf, tablename); Scan s = new Scan(); SingleColumnValueFilter sf = new SingleColumnValueFilter(Bytes.toBytes(strF), Bytes.toBytes(strC), CompareOp.EQUAL, Bytes.toBytes(strClass)); s.setFilter(sf); ResultScanner rs = table.getScanner(s); HashMap<String, Integer> hashMap = new HashMap<>(); int i=0; for (Result r : rs) { i=i+1; byte[] row = r.getRow(); byte[] value = r.getValue(Bytes.toBytes(strF), Bytes.toBytes(strC)); byte[] value2 = r.getValue(Bytes.toBytes(strF), Bytes.toBytes("class")); if(!hashMap.containsKey(Bytes.toString(value2))) { //集合中没有该单词,值定义为1 hashMap.put(Bytes.toString(value2), 1); }else if(hashMap.containsKey(Bytes.toString(value2))) { //集合中有该单词,值+1;key不变; int b=hashMap.get(Bytes.toString(value2)); hashMap.put(Bytes.toString(value2), b+1); } // System.out.println("Filter: " + Bytes.toString(row) + " is in " + strC + " " + Bytes.toString(value)+", class " + Bytes.toString(value2)); } System.out.println(hashMap); System.out.println("i="+i); rs.close(); table.close(); } public void get(String tablename,String row,String info,String name) throws Exception{ HTable table=new HTable(conf,tablename); Get g=new Get(Bytes.toBytes(row)); Result result = table.get(g); byte[] val = result.getValue(Bytes.toBytes(info),Bytes.toBytes(name)); System.out.println(info+" "+name+" "+"Values =" + Bytes.toString(val)); } public void scan(String tablename,String cf,String column) throws Exception{ HTable table=new HTable(conf,tablename); Scan s=new Scan(); s.setStartRow(Bytes.toBytes("0")); s.setStopRow(Bytes.toBytes("g")); // s.addColumn(Bytes.toBytes("info"), Bytes.toBytes("name")); ResultScanner res=table.getScanner(s); for(Result r:res) { byte[] row=r.getRow(); byte[] val=r.getValue(Bytes.toBytes(cf), Bytes.toBytes(column)); System.out.println("Scan:"+Bytes.toString(row) +" values is "+Bytes.toString(val)); } res.close(); table.close(); } public void delete(String tablename,String row,String cf,String column,String data) throws Exception{ HTable table=new HTable(conf,tablename); List<Delete> ls=new ArrayList<Delete>(); Delete p=new Delete(Bytes.toBytes(row)); ls.add(p); //p.add(Bytes.toBytes(cf), Bytes.toBytes(column), Bytes.toBytes(data)); table.delete(ls); table.close(); } }
插进表里面这种数据貌似是对的
或者直接用spark简单粗暴过滤这个，这个是这一句直接来的
先hbase主函数这个：
ho.filterBySingleColumnValueFilter(tab, "log","level","ERROR");
会有一个数据产生，把数据拷贝到一个文件里面，再spark单词计数哈哈哈
方法类这样： public void filterBySingleColumnValueFilter1(String tablename, String strF, String strC, String strClass) throws Exception { HTable table = new HTable(conf, tablename); Scan s = new Scan(); SingleColumnValueFilter sf = new SingleColumnValueFilter(Bytes.toBytes(strF), Bytes.toBytes(strC), CompareOp.EQUAL, Bytes.toBytes(strClass)); s.setFilter(sf); ResultScanner rs = table.getScanner(s); for (Result r : rs) { byte[] row = r.getRow(); byte[] value = r.getValue(Bytes.toBytes(strF), Bytes.toBytes(strC)); byte[] value2 = r.getValue(Bytes.toBytes(strF), Bytes.toBytes("class")); System.out.println("Filter: " + Bytes.toString(row) + " is in " + strC + " " + Bytes.toString(value)+", class " + Bytes.toString(value2)); } rs.close(); table.close(); }
package sparkj; import java.text.SimpleDateFormat; import java.util.Arrays; import java.util.regex.Matcher; import java.util.regex.Pattern; import org.apache.spark.SparkConf; import org.apache.spark.api.java.JavaPairRDD; import org.apache.spark.api.java.JavaRDD; import org.apache.spark.api.java.JavaSparkContext; import scala.Tuple2; public class sp { public static void main(String[] args) { // TODO Auto-generated method stub SparkConf sparkConf = new SparkConf().setAppName("PeopleInfoCalculator").setMaster("local[2]"); JavaSparkContext sc = new JavaSparkContext(sparkConf); JavaRDD<String> filedata=sc.textFile("file:///root/555.txt"); // JavaRDD<String> data2=filedata.filter(f->f.matches("[0-9].*")) // .map(f->String2Date(f.split(" ")[0],f.split(" ")[1])+" "+f.split(" ")[2]+" "+f.split(" ")[3].split(":")[0]); // // data2.foreach(f->System.err.println(f)); // data2.saveAsTextFile("file:///root/555copy.txt"); JavaPairRDD<String,Integer> rdd6=filedata.mapToPair(f->new Tuple2<>(f.split(",")[1],1)); JavaPairRDD<String,Integer> rdd7=rdd6.reduceByKey((x,y)->x+y); rdd7.foreach(f->System.err.println(f)); } public static long String2Date(String date,String time) throws Exception{ String newdate = date + " " + time; SimpleDateFormat s = new SimpleDateFormat("yyyy-MM-dd hh:mm:ss,SSS"); return s.parse(newdate).getTime(); } }
最后结果
