先启动虚拟机，并通过Xshell工具进行连接连接成功后，进入到opt目录创建一个software文件夹然后进入到software中通过Xftp上传jdk安装包，上传步骤省略注意：安装hadoop之前先安装java因为这个文件是以tar.gz结尾，它是一个压缩文件，我们需要对这个压缩文件进行解压。解压完成之后，我们可以看见我们就有了这个解压好之后的文件夹然后我们配置jdk的环境变量，输入vim /etc/profile注意：如果提示下面这个代表我们当前Linux系统中没有安装vim软件，所以导致命令无法找到。我们需要执行yum install vim 回车进行安装。重新输入vim /etc/profile进入到这个界面之后，按方向键下键，来到最后一行检查输入法是否是英文，如果不是请先切换，然后按一下 i 键进入编辑模式，多按几次回车然后在底部写入，一定要注意自己的路径，你可能路径和我不要一样需要修改输入完成之后，按一下esc退出编辑模式，然后输入:wq 进行保存，如果你觉得你输入错误不想保存请输入:q!更新环境变量 source /etc/profile检验Java环境变量是否配置正确然后在相同的目录下，上传hadoop的压缩包对这个压缩包进行解压解压完成后，我们可以看见一个这样的文件夹我们需要进入到这个文件夹下的配置文件夹目录分别修改以下5个配置文件，都使用vim文件编辑器修改，修改后退出保存后再继续下一个文件的修改vim hadoop-env.shvim core-site.xmlvim hdfs-site.xml修改第四个文件时，需要拷贝一个模板文件后再编辑cp mapred-site.xml.template mapred-site.xmlvim mapred-site.xmlvim yarn-site.xml上面5个配置好了之后，继续修改配置文件vim /etc/profile，然后在java的环境变量底下添加hadoop的环境变量设置好了，退出并保存，再更新一下环境变量配置SSH免密登录，输入ssh-keygen后连续按三次回车然后将这个生成的密钥拷贝到此主机，ssh-copy-id root@127.0.0.1然后格式化Hadoop出现success代表格式化成功最后启动Hadoop，start-all.sh。第一次如果出现yes要输入yes输出jps进行校验，如果出现下面几个节点，就代表hadoop安装成功了
第一关：SQL API使用import org.apache.flink.api.java.DataSet; import org.apache.flink.api.java.ExecutionEnvironment; import org.apache.flink.api.java.operators.DataSource; import org.apache.flink.core.fs.FileSystem; import org.apache.flink.table.api.Table; import org.apache.flink.table.api.java.BatchTableEnvironment; public class Step1 { public static void main(String[] args) throws Exception { // 请在此处编写代码 /********* Begin *********/ // 创建Flink运行环境 ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); // 创建Table环境 BatchTableEnvironment batchTableEnvironment = BatchTableEnvironment.create(env); // 读取csv文件创建DataSource DataSource<Player> platerDataSource = env.readCsvFile("/root/score.csv") .fieldDelimiter(",") .pojoType(Player.class, "sj", "player", "cc", "sf", "time", "zg", "qd", "gm", "score"); // 创建表对象 Table table = batchTableEnvironment.fromDataSet(platerDataSource); batchTableEnvironment.registerTable("mytable",table); // 查询 Table table1 = batchTableEnvironment.sqlQuery("select player, round(sum(score),1) as sum_score FROM mytable GROUP BY player ORDER BY sum_score desc LIMIT 3"); // 将 Table 转换成 DataSet 输出 DataSet<Result> playerDataDataSet = batchTableEnvironment.toDataSet(table1, Result.class); playerDataDataSet.print(); /********* End *********/ } // 定义Result实例对象 public static class Result { public String player; public double sum_score; //重写toString方法 @Override public String toString() { return "Result{" + "player='" + player + '\'' + ", sum_score=" + sum_score + '}'; } } // 定义球员实例对象 public static class Player { private String sj; // 赛季 private String player; // 球员 private int cc; // 出场 private int sf; // 首发 private double time; // 时间 private double zg; // 助攻 private double qd; // 抢断 private double gm; // 盖帽 private double score; // 得分 public String getSj() { return sj; } public void setSj(String sj) { this.sj = sj; } public String getPlayer() { return player; } public void setPlayer(String player) { this.player = player; } public int getCc() { return cc; } public void setCc(int cc) { this.cc = cc; } public int getSf() { return sf; } public void setSf(int sf) { this.sf = sf; } public double getTime() { return time; } public void setTime(double time) { this.time = time; } public double getZg() { return zg; } public void setZg(double zg) { this.zg = zg; } public double getQd() { return qd; } public void setQd(double qd) { this.qd = qd; } public double getGm() { return gm; } public void setGm(double gm) { this.gm = gm; } public double getScore() { return score; } public void setScore(double score) { this.score = score; } // 重写toString方法 @Override public String toString() { return "Player{" + "sj='" + sj + '\'' + ", player='" + player + '\'' + ",cc='" + cc + '\'' + ",sf='" + sf + '\'' + ",time='" + time + '\'' + ",zg='" + zg + '\'' + ",qd='" + qd + '\'' + ",gm='" + gm + '\'' + ",score='" + score + '\'' + '}'; } } }第二关：Table API使用import org.apache.flink.api.java.DataSet; import org.apache.flink.api.java.ExecutionEnvironment; import org.apache.flink.api.java.operators.DataSource; import org.apache.flink.core.fs.FileSystem; import org.apache.flink.table.api.Table; import org.apache.flink.table.api.GroupedTable; import org.apache.flink.table.api.java.BatchTableEnvironment; public class Step2 { public static void main(String[] args) throws Exception { // 请在此处编写代码 /********* Begin *********/ // 创建Flink运行环境 ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); // 创建Table环境 BatchTableEnvironment batchTableEnvironment = BatchTableEnvironment.create(env); // 读取csv文件创建DataSource DataSource<Player> platerDataSource = env.readCsvFile("/root/score.csv") .fieldDelimiter(",") .pojoType(Player.class, "sj", "player", "cc", "sf", "time", "zg", "qd", "gm", "score"); // 创建表对象 Table table = batchTableEnvironment.fromDataSet(platerDataSource); batchTableEnvironment.registerTable("mytable",table); // 查询 //Table table1 = batchTableEnvironment.sqlQuery("select player, round(sum(score),1) as sum_score FROM mytable GROUP BY player ORDER BY sum_score desc LIMIT 3"); Table mytable = batchTableEnvironment.scan("mytable"); // 扫描表 Table filter = mytable.filter("score>32"); // 过滤出 score 大于 32 的球员 GroupedTable player = filter.groupBy("player"); // 对球员分组 Table select = player.select("player"); // 查询出 player 列 // 将 Table 转换成 DataSet 输出 DataSet<Result> playerDataDataSet = batchTableEnvironment.toDataSet(select, Result.class); playerDataDataSet.print(); /********* End *********/ } // 定义Result实例对象 public static class Result { public String player; //重写toString方法 @Override public String toString() { return "Result{" + "player='" + player + '\'' + '}'; } } // 定义球员实例对象 public static class Player { private String sj; // 赛季 private String player; // 球员 private int cc; // 出场 private int sf; // 首发 private double time; // 时间 private double zg; // 助攻 private double qd; // 抢断 private double gm; // 盖帽 private double score; // 得分 public String getSj() { return sj; } public void setSj(String sj) { this.sj = sj; } public String getPlayer() { return player; } public void setPlayer(String player) { this.player = player; } public int getCc() { return cc; } public void setCc(int cc) { this.cc = cc; } public int getSf() { return sf; } public void setSf(int sf) { this.sf = sf; } public double getTime() { return time; } public void setTime(double time) { this.time = time; } public double getZg() { return zg; } public void setZg(double zg) { this.zg = zg; } public double getQd() { return qd; } public void setQd(double qd) { this.qd = qd; } public double getGm() { return gm; } public void setGm(double gm) { this.gm = gm; } public double getScore() { return score; } public void setScore(double score) { this.score = score; } // 重写toString方法 @Override public String toString() { return "Player{" + "sj='" + sj + '\'' + ", player='" + player + '\'' + ",cc='" + cc + '\'' + ",sf='" + sf + '\'' + ",time='" + time + '\'' + ",zg='" + zg + '\'' + ",qd='" + qd + '\'' + ",gm='" + gm + '\'' + ",score='" + score + '\'' + '}'; } } }第三关：用户自定义函数UDF使用import org.apache.flink.api.java.DataSet; import org.apache.flink.api.java.ExecutionEnvironment; import org.apache.flink.api.java.operators.DataSource; import org.apache.flink.core.fs.FileSystem; import org.apache.flink.table.api.Table; import org.apache.flink.table.api.java.BatchTableEnvironment; import org.apache.flink.table.functions.ScalarFunction; import java.text.SimpleDateFormat; import java.util.Date; public class Step3 { public static void main(String[] args) throws Exception { // 创建批处理运行环境 ExecutionEnvironment environment = ExecutionEnvironment.getExecutionEnvironment(); // 创建表环境 BatchTableEnvironment tableEnvironment = BatchTableEnvironment.create(environment); // 读取数据 DataSource<taxi> dataSource = environment.readCsvFile("/root/data.csv").fieldDelimiter("\t"). ignoreFirstLine().pojoType(taxi.class, "TRIP_ID", "CALL_TYPE", "ORIGIN_CALL", "TAXI_ID", "ORIGIN_STAND", "_TIMESTAMP", "POLYLINE"); // 定义表对象 Table table = tableEnvironment.fromDataSet(dataSource); // 请在此处编写代码 /********* Begin *********/ // 注册自定义函数 tableEnvironment.registerFunction("DataTimeFormat",new DataTimeFormat()); // 注册表 tableEnvironment.registerTable("mytable",table); // 使用自定义函数查询 Table table1 = tableEnvironment.sqlQuery("select TRIP_ID,CALL_TYPE,DataTimeFormat(_TIMESTAMP) as FORMATTIME from mytable order by FORMATTIME limit 5"); // 将Table 转化为 DataSet DataSet<Result> resultDataSet = tableEnvironment.toDataSet(table1, Result.class); resultDataSet.print(); /********* End *********/ } // 自定义DataTimeFormat函数，对日期进行格式化 public static class DataTimeFormat extends ScalarFunction { // 需要格式化的日期转化规则 public SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); public String eval(String _TIMESTAMP) { // 返回格式化之后的数据 return simpleDateFormat.format(new Date(Long.parseLong(_TIMESTAMP))); } } // 定义Result实例对象 public static class Result { public String TRIP_ID; public String CALL_TYPE; public String FORMATTIME; //重写toString方法 @Override public String toString() { return "Result{" + "TRIP_ID='" + TRIP_ID + '\'' + ", CALL_TYPE=" + '\'' + CALL_TYPE + '\'' + ", FORMATTIME=" + '\'' + FORMATTIME + '\'' + '}'; } } // taxi 实例 public static class taxi { public String TRIP_ID; public String CALL_TYPE; public String ORIGIN_CALL; public String TAXI_ID; public String ORIGIN_STAND; public String _TIMESTAMP; public String POLYLINE; @Override public String toString() { return "taxi{" + "TRIP_ID='" + TRIP_ID + '\'' + ", CALL_TYPE='" + CALL_TYPE + '\'' + ", ORIGIN_CALL='" + ORIGIN_CALL + '\'' + ", TAXI_ID='" + TAXI_ID + '\'' + ", ORIGIN_STAND='" + ORIGIN_STAND + '\'' + ", TIMESTAMP='" + _TIMESTAMP + '\'' + ", POLYLINE='" + POLYLINE + '\'' + '}'; } } } 第4关：用户自定义函数 UDTF 的使用import org.apache.flink.api.java.DataSet; import org.apache.flink.api.java.ExecutionEnvironment; import org.apache.flink.api.java.operators.DataSource; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.table.api.Table; import org.apache.flink.table.api.java.BatchTableEnvironment; import org.apache.flink.table.functions.TableFunction; public class FlinkTest { public static void main(String[] args) throws Exception { // 创建 Flink 运行环境（批处理环境） ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment(); // 创建表环境 BatchTableEnvironment batchTableEnvironment = BatchTableEnvironment.create(env); // 读取 csv 文件创建Dataset DataSource<Student> studentDataSource = env.readCsvFile("/root/a.csv").includeFields(true, true, true).fieldDelimiter(",").ignoreFirstLine().pojoType(Student.class, "name", "bir", "score"); // 创建表对象 Table table = batchTableEnvironment.fromDataSet(studentDataSource); // 请在此处编写代码 /********* Begin *********/ // 注册自定义函数 batchTableEnvironment.registerFunction("splitFun",new SplitFun()); // 注册表 batchTableEnvironment.registerTable("mytable",table); // 使用自定义函数查询 Table table1 = batchTableEnvironment.sqlQuery("select name,dat,tim from mytable,LATERAL TABLE(splitFun(bir)) as T(dat, tim)"); // 将Table转化为DataSet DataSet<Result> resultDataSet = batchTableEnvironment.toDataSet(table1, Result.class); resultDataSet.print(); /********* End *********/ } // 自定义函数，将出生日期进行拆分 public static class SplitFun extends TableFunction<Tuple2<String, String>> { // 通过用什么进行拆分的构造函数 public SplitFun() { } // 用于构造方法进行传参 public SplitFun(String separator) { this.separator = separator; } private String separator = " "; public void eval(String str) { String[] datas = str.split(separator); // 对数据进行拆分 // 获取拆分后前面的数据和后面的数据 collect(new Tuple2<String, String>(datas[0], datas[1])); } } // 返回结果 public static class Result{ public String name; public String dat; public String tim; @Override public String toString() { return " Result{" + "name='" + name + '\'' + ", dat=" + dat + ", tim=" + tim + '}'; } } public static class Student{ private String name; private int score; private String bir; public int getScore() { return score; } public void setScore(int score) { this.score = score; } public String getName() { return name; } public void setName(String name) { this.name = name; } public String getBir() { return bir; } public void setBir(String addr) { this.bir = addr; } @Override public String toString() { return "Student{" + "name='" + name + '\'' + ", score='" + score + '\'' + ",addr='" + bir + '\'' + '}'; } } }
感谢宋同学提供的下面内容基于物品推荐算法import org.apache.log4j.{Level, Logger} import org.apache.spark.mllib.linalg.SparseVector import org.apache.spark.mllib.linalg.distributed.{CoordinateMatrix, IndexedRow, MatrixEntry, RowMatrix} import org.apache.spark.rdd.RDD import org.apache.spark.{SparkConf, SparkContext} object ItemBasedCF { def main(args: Array[String]): Unit = { Logger.getLogger("org.apache.spark").setLevel(Level.ERROR) Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF) //读入数据 val conf = new SparkConf().setAppName("ItemBasedCFModel").setMaster("local") val sc = new SparkContext(conf) val data = sc.textFile("/root/data/als/ratingdata.txt") /*MatrixEntry代表一个分布式矩阵中的每一行(Entry) * 这里的每一项都是一个(i: Long, j: Long, value: Double) 指示行列值的元组tuple。 * 其中i是行坐标，j是列坐标，value是值。*/ val parseData: RDD[MatrixEntry] = data.map(_.split(",") match { case Array(user, item, rate) => MatrixEntry(user.toLong, item.toLong, rate.toDouble)}) data.map(_.split(",") match { case Array(user, item, rate) => MatrixEntry(user.toLong, item.toLong, rate.toDouble) }) //CoordinateMatrix是Spark MLLib中专门保存user_item_rating这种数据样本的 val ratings = new CoordinateMatrix(parseData) /* 由于CoordinateMatrix没有columnSimilarities方法，所以我们需要将其转换成RowMatrix矩阵，调用他的columnSimilarities计算其相似性 * RowMatrix的方法columnSimilarities是计算，列与列的相似度，现在是user_item_rating，与基于用户的CF不同的是，这里不需要进行矩阵的转置，直接就是物品的相似*/ val matrix: RowMatrix = ratings.toRowMatrix() //需求：为某一个用户推荐商品。基本的逻辑是：首先得到某个用户评价过（买过）的商品，然后计算其他商品与该商品的相似度，并排序；从高到低，把不在用户评价过 //商品里的其他商品推荐给用户。 //例如：为用户2推荐商品 //第一步：得到用户2评价过（买过）的商品 take(5)表示取出所有的5个用户 2:表示第二个用户 //解释：SparseVector：稀疏矩阵 val user2pred = matrix.rows.take(5)(2) val prefs: SparseVector = user2pred.asInstanceOf[SparseVector] val uitems = prefs.indices //得到了用户2评价过（买过）的商品的ID val ipi = (uitems zip prefs.values) //得到了用户2评价过（买过）的商品的ID和评分，即：(物品ID,评分) //计算物品的相似性，并输出 val similarities = matrix.columnSimilarities() val indexdsimilar = similarities.toIndexedRowMatrix().rows.map { case IndexedRow(idx, vector) => (idx.toInt, vector) } //ij表示：其他用户购买的商品与用户2购买的该商品的相似度 val ij = sc.parallelize(ipi).join(indexdsimilar).flatMap { case (i, (pi, vector: SparseVector)) => (vector.indices zip vector.values) } /********** begin **********/ //ij1表示：其他用户购买过，但不在用户2购买的商品的列表中的商品和评分 val ij1 = ij.filter { case (item, pref) => !uitems.contains(item) } //将这些商品的评分求和，并降序排列，并推荐前两个物品 val ij2 = ij1.reduceByKey(_ + _).sortBy(_._2, false).take(2) /********** end **********/ // crgjl //取消以下1行注释 for (id <- ij2) print(id._1 + " ") sc.stop() } } 基于用户的推荐算法import org.apache.log4j.{Level, Logger} import org.apache.spark.mllib.linalg.distributed.{CoordinateMatrix, MatrixEntry, RowMatrix} import org.apache.spark.rdd.RDD import org.apache.spark.{SparkConf, SparkContext} object UserBasedCF { def main(args: Array[String]): Unit = { Logger.getLogger("org.apache.spark").setLevel(Level.ERROR) Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF) // 创建一个SparkContext val conf = new SparkConf().setAppName("UserBasedCF").setMaster("local") val sc = new SparkContext(conf) // 读入数据 val data = sc.textFile("/root/data/als/ratingdata.txt") // 解析出评分矩阵的每一行 val parseData: RDD[MatrixEntry] = data.map(_.split(",") match { case Array(user, item, rate) => MatrixEntry(user.toLong, item.toLong, rate.toDouble) }) // 构建关联矩阵 val ratings = new CoordinateMatrix(parseData) // 转置矩阵以计算列(用户)的相似性 val matrix: RowMatrix = ratings.transpose().toRowMatrix() // 计算得到用户的相似度矩阵 val similarities = matrix.columnSimilarities() // 得到某个用户对所有物品的评分 val ratingOfUser1 = ratings.entries.filter(_.i == 1). map(x => (x.j, x.value)). sortBy(_._1). map(_._1). collect(). toList. toArray // 得到用户1相对于其他用户的相似性 val similarityOfUser1 = similarities.entries.filter(_.i == 1). sortBy(_.value, false). map(_.value). collect // 需求：为用户1推荐2个商品 // 思路：找到与用户1相似性最高的两个用户，将这两个用户评过分的物品，用户1没有评过分的物品推荐给用户1 val similarityTopUser = similarities.entries.filter(_.i == 1). sortBy(_.value, false). map(x => (x.j, x.value)). collect. take(2) //println("与用户1最相似的两个用户如下：") for (s <- similarityTopUser) print(s._1 + " ") for (s <- similarityTopUser) { // 找到这两个用户评过分的商品，与用户1没有评过分的物品 val userId = s._1 val ratingOfTemp = ratings.entries.filter(_.i == userId). map(x => (x.j, x.value)). sortBy(_._1). map(_._1). collect(). toList. toArray // 用户1与当前用户求差集 val dis = ratingOfTemp diff ratingOfUser1 //println("用户" + userId + "要推荐给用户1的商品id为") for (id <- dis) print(id + " ") } sc.stop() } } 基于ALS的推荐算法import org.apache.spark.ml.evaluation.RegressionEvaluator import org.apache.spark.ml.recommendation.ALS import org.apache.spark.sql.{DataFrame, SparkSession} object ALS { case class Rating(userId: Int, movieId: Int, rating: Float) def parseRating(str: String): Rating = { val fields = str.split(",") assert(fields.size == 3) Rating(fields(0).toInt, fields(1).toInt, fields(2).toFloat) } def main(args: Array[String]) { val spark = SparkSession .builder .master("local") .appName("ALS") .getOrCreate() import spark.implicits._ val ratings = spark.read.textFile("data/als/ratingdata.txt") .map(parseRating) .toDF() val Array(training, test) = ratings.randomSplit(Array(0.8, 0.2)) // Build the recommendation model using ALS on the training data val als = new ALS() .setMaxIter(5) .setRegParam(0.01) .setUserCol("userId") .setItemCol("movieId") .setRatingCol("rating") val model = als.fit(training) // Evaluate the model by computing the RMSE on the test data // Note we set cold start strategy to 'drop' to ensure we don't get NaN evaluation metrics model.setColdStartStrategy("drop") val predictions = model.transform(test) evaluatingRMSE(predictions) spark.stop() } def evaluatingRMSE(predictions:DataFrame):Unit = { val evaluator = new RegressionEvaluator() .setMetricName("rmse") .setLabelCol("rating") .setPredictionCol("prediction") val rmse = evaluator.evaluate(predictions) if (rmse < 2){ print("\n" + "good") }else{ println() predictions.show(false) } } } 基于随机森林预测贷款风险import org.apache.spark.rdd.RDD import org.apache.spark.sql.{DataFrame,SparkSession} import org.apache.spark.ml.classification.RandomForestClassifier import org.apache.spark.ml.evaluation.BinaryClassificationEvaluator import org.apache.spark.ml.feature.StringIndexer import org.apache.spark.ml.feature.VectorAssembler import org.apache.spark.ml.tuning.{CrossValidator, ParamGridBuilder} import org.apache.spark.ml.{Pipeline, PipelineStage} import org.apache.spark.mllib.evaluation.RegressionMetrics object Credit { case class Credit( creditability: Double, balance: Double, duration: Double, history: Double, purpose: Double, amount: Double, savings: Double, employment: Double, instPercent: Double, sexMarried: Double, guarantors: Double, residenceDuration: Double, assets: Double, age: Double, concCredit: Double, apartment: Double, credits: Double, occupation: Double, dependents: Double, hasPhone: Double, foreign: Double ) def parseCredit(line: Array[Double]): Credit = { Credit( line(0), line(1) - 1, line(2), line(3), line(4), line(5), line(6) - 1, line(7) - 1, line(8), line(9) - 1, line(10) - 1, line(11) - 1, line(12) - 1, line(13), line(14) - 1, line(15) - 1, line(16) - 1, line(17) - 1, line(18) - 1, line(19) - 1, line(20) - 1 ) } def parseRDD(rdd: RDD[String]): RDD[Array[Double]] = { rdd.map(_.split(",")).map(_.map(_.toDouble)) } def evaluatingAUC(predictedResultDF:DataFrame, labelstring:String):Unit = { val evaluator = new BinaryClassificationEvaluator().setLabelCol(labelstring) val predictionAUC = evaluator.setMetricName("areaUnderROC").evaluate(predictedResultDF) if(predictionAUC > 0.6){ print("\n" + "good") }else{ print(s"areaUnderROC: $predictionAUC") } } def main(args: Array[String]) { val spark = SparkSession .builder .appName("Credit") .master("local") .getOrCreate() import spark.implicits._ val creditDF = parseRDD(spark.sparkContext.textFile("/root/data/germancredit.csv")).map(parseCredit).toDF() creditDF.createTempView("credit") val featureCols = Array("balance", "duration", "history", "purpose", "amount", "savings", "employment", "instPercent", "sexMarried", "guarantors", "residenceDuration", "assets", "age", "concCredit", "apartment", "credits", "occupation", "dependents", "hasPhone", "foreign") val assembler = new VectorAssembler().setInputCols(featureCols).setOutputCol("features") val df2 = assembler.transform(creditDF) val labelIndexer = new StringIndexer().setInputCol("creditability").setOutputCol("label") val df3 = labelIndexer.fit(df2).transform(df2) val splitSeed = 5043 val Array(trainingData, testData) = df3.randomSplit(Array(0.7, 0.3), splitSeed) val classifier = new RandomForestClassifier().setImpurity("gini").setMaxDepth(5).setNumTrees(20).setFeatureSubsetStrategy("auto").setSeed(5043) val model = classifier.fit(trainingData) val predictions = model.transform(testData) evaluatingAUC(predictions,"label") spark.stop() } } 基于多层感知器的手机短信分类import org.apache.spark.ml.Pipeline import org.apache.spark.ml.classification.MultilayerPerceptronClassifier import org.apache.spark.ml.evaluation.BinaryClassificationEvaluator import org.apache.spark.ml.feature.{IndexToString, StringIndexer, Word2Vec} import org.apache.spark.sql.{DataFrame,SparkSession} object SMSClassifier { final val VECTOR_SIZE = 100 def evaluatingAUC(predictedResultDF:DataFrame, labelcol: String):Unit = { val evaluator = new BinaryClassificationEvaluator().setLabelCol(labelcol).setRawPredictionCol("prediction") val predictionAUC = evaluator.setMetricName("areaUnderROC").evaluate(predictedResultDF) if(predictionAUC > 0.8){ print("\n" + "good") }else{ print(s"areaUnderROC: $predictionAUC") } } def main(args: Array[String]) { val spark = SparkSession .builder .master("local") .appName("SMS Message Classification (HAM or SPAM)") .getOrCreate() val parsedRDD = spark.sparkContext.textFile("data/smsspamcollection/SMSSpamCollection").map(_.split("\t")).map(eachRow => { (eachRow(0),eachRow(1).split(" ")) }) val msgDF = spark.createDataFrame(parsedRDD).toDF("label","message") val labelIndexer = new StringIndexer() .setInputCol("label") .setOutputCol("indexedLabel") .fit(msgDF) val word2Vec = new Word2Vec() .setInputCol("message") .setOutputCol("features") .setVectorSize(VECTOR_SIZE) .setMinCount(1) val layers = Array[Int](VECTOR_SIZE,6,5,2) val mlpc = new MultilayerPerceptronClassifier() .setLayers(layers) .setBlockSize(512) .setSeed(1234L) .setMaxIter(128) .setFeaturesCol("features") .setLabelCol("indexedLabel") .setPredictionCol("prediction") val labelConverter = new IndexToString() .setInputCol("prediction") .setOutputCol("predictedLabel") .setLabels(labelIndexer.labels) val Array(trainingData, testData) = msgDF.randomSplit(Array(0.8, 0.2)) val pipeline = new Pipeline().setStages(Array(labelIndexer,word2Vec,mlpc,labelConverter)) val model = pipeline.fit(trainingData) val predictionResultDF = model.transform(testData) evaluatingAUC(predictionResultDF,"indexedLabel") spark.stop() } }
注意，我们需要编写3个Java文件第一个文件:DBHelper.java(数据库工具类)package com; import java.sql.Connection; import java.sql.DriverManager; import java.sql.SQLException; public class DBHelper { /********** begin **********/ // mysql驱动 private static final String driver = "com.mysql.jdbc.Driver"; // mysql连接地址 private static final String url = "jdbc:mysql://localhost:3306/mydb?useUnicode=true&characterEncoding=UTF-8"; // mysql的用户名 private static final String username = "root"; // mysql的密码 private static final String password = "123123"; // 连接对象 private static Connection conn = null; // 设置静态变量，用于加载数据库驱动 static { try { // 加载数据库驱动 Class.forName(driver); } catch (Exception e) { // 抛出异常 e.printStackTrace(); } } // 创建连接类，连接数据库 public static Connection getConn() { // 如果对象不存在，开始创建 if (conn == null) { try { // 创建连接对象 conn = DriverManager.getConnection(url, username, password); } catch (SQLException e) { // 如果无法创建数据库连接对象，则抛出数据库异常 e.printStackTrace(); } // 创建成功后返回这个对象 return conn; } // 如果对象存在，直接返回连接对象 return conn; } public static void main(String[] args) { // 通过数据库连接工具类进行连接 Connection conn = DBHelper.getConn(); } /********** end **********/ } 第二文件：JsonMap.java(MapReduce的操作类)package com; import com.alibaba.fastjson.JSONObject; import org.apache.hadoop.hbase.client.Put; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; import java.io.IOException; import java.sql.Connection; import java.sql.ResultSet; import java.sql.SQLException; import java.sql.Statement; import java.util.HashMap; import java.util.Map; public class JsonMap extends Mapper<LongWritable, Text, NullWritable, Put> { /********** begin **********/ // 用户存放城市编码和城市名字的数据 Map<String, String> map = new HashMap<String, String>(); Put put; @Override protected void setup(Context context) throws IOException, InterruptedException { // 通过数据库工具类获取对象 Connection conn = DBHelper.getConn(); try { // 查询数据 Statement s = conn.createStatement(); String sql = "select * from province"; ResultSet r = s.executeQuery(sql); // 循环结果 while (r.next()) { // 获取城市编码 String cityCode = r.getString(1); // 获取城市的名字 String cityName = r.getString(2); // 将获取到的数据放入map集合中 // 前面是键，后面是值 map.put(cityCode, cityName); } } catch (SQLException e) { // 如果数据异常时，抛出该异常 e.printStackTrace(); } } // 编写map类 public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { // 获取数据 String line = value.toString(); // 开始解析Json数据 JSONObject j = JSONObject.parseObject(line); // 因为要解析的数据有14条，用一个数组存放这个14条数据 String[] data = new String[14]; data[0] = j.getString("id"); data[1] = j.getString("company_name"); data[2] = j.getString("eduLevel_name"); data[3] = j.getString("emplType"); data[4] = j.getString("jobName"); // 对工资进行解析 String salary = j.getString("salary"); if (salary.contains("K-")) { Double a = Double.valueOf(salary.substring(0, salary.indexOf("K"))); Double b = Double.valueOf(salary.substring(salary.indexOf("-") + 1, salary.lastIndexOf("K"))); data[5] = (a + b) / 2 + ""; } else { data[5] = "0"; } data[6] = j.getString("createDate"); data[7] = j.getString("endDate"); data[8] = j.getString("city_code"); data[9] = j.getString("companySize"); data[10] = j.getString("welfare"); data[11] = j.getString("responsibility"); data[12] = j.getString("place"); data[13] = j.getString("workingExp"); // 检查这些数据是否为空 for (String s : data) { if (s.equals("") || s == null) { // 如果为空直接返回，不执行下面的操作 return; } } String info = "info"; // 将所有数据添加进info中 put = new Put(data[0].getBytes()); put.addColumn(info.getBytes(), "company_name".getBytes(), data[1].getBytes()); put.addColumn(info.getBytes(), "eduLevel_name".getBytes(), data[2].getBytes()); put.addColumn(info.getBytes(), "emplType".getBytes(), data[3].getBytes()); put.addColumn(info.getBytes(), "jobName".getBytes(), data[4].getBytes()); put.addColumn(info.getBytes(), "salary".getBytes(), data[5].getBytes()); put.addColumn(info.getBytes(), "createDate".getBytes(), data[6].getBytes()); put.addColumn(info.getBytes(), "endDate".getBytes(), data[7].getBytes()); put.addColumn(info.getBytes(), "city_name".getBytes(), data[8].getBytes()); put.addColumn(info.getBytes(), "companySize".getBytes(), data[9].getBytes()); put.addColumn(info.getBytes(), "welfare".getBytes(), data[10].getBytes()); put.addColumn(info.getBytes(), "responsibility".getBytes(), data[11].getBytes()); put.addColumn(info.getBytes(), "place".getBytes(), data[12].getBytes()); put.addColumn(info.getBytes(), "workingExp".getBytes(), data[13].getBytes()); // 将它写入到HDFS中 context.write(NullWritable.get(), put); } /********** end **********/ } 第三个文件：JsonTest.class(测试操作类)package com; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.TableName; import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor; import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder; import org.apache.hadoop.hbase.client.Connection; import org.apache.hadoop.hbase.client.TableDescriptorBuilder; import org.apache.hadoop.hbase.client.Admin; import org.apache.hadoop.hbase.client.ConnectionFactory; import org.apache.hadoop.hbase.util.Bytes; import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil; public class JsonTest { public static void main(String[] args) throws Exception { Configuration config = HBaseConfiguration.create(); // 设置zookeeper的配置 config.set("hbase.zookeeper.quorum", "127.0.0.1"); Connection connection = ConnectionFactory.createConnection(config); Admin admin = connection.getAdmin(); TableName tableName = TableName.valueOf("job"); boolean isExists = admin.tableExists(tableName); if (!isExists) { TableDescriptorBuilder tableDescriptor = TableDescriptorBuilder.newBuilder(tableName); ColumnFamilyDescriptor family = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes("info")).build();// 构建列族对象 tableDescriptor.setColumnFamily(family); // 设置列族 admin.createTable(tableDescriptor.build()); // 创建表 } else { admin.disableTable(tableName); admin.deleteTable(tableName); TableDescriptorBuilder tableDescriptor = TableDescriptorBuilder.newBuilder(tableName); ColumnFamilyDescriptor family = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes("info")).build();// 构建列族对象 tableDescriptor.setColumnFamily(family); // 设置列族 admin.createTable(tableDescriptor.build()); // 创建表 } /********** begin **********/ // 设置需要执行任务 Job job = Job.getInstance(config); // 设置map的执行任务 job.setJarByClass(JsonTest.class); job.setMapperClass(JsonMap.class); job.setMapOutputKeyClass(NullWritable.class); // 因为该任务没有用到reduce，所以需要设置其值为0 job.setNumReduceTasks(0); // 设置数据的所在位置 FileInputFormat.setInputPaths(job, new Path("/root/data/data.json")); // 开始执行任务 TableMapReduceUtil.initTableReducerJob("job", null, job); TableMapReduceUtil.addDependencyJars(job); // 提交任务 boolean b = job.waitForCompletion(true); // 退出任务 System.exit(b ? 0 : 1); /********** end **********/ } } 上面编写完代码后，我们需要到命令行去启动Hbase启动hbase（start-hbase.sh）完成后，我们就可以开始测评啦！
第一关：创建数据库首先我们需要来到 命令行中，输入 mongo进入到mongodb数据库中我们再看看题目的要求：在右侧命令行中连接 MongoDB ，创建一个名为 mydb 的数据库，并插入数据：_id:1,name:”李小红”我们首先需要 使用mydb数据库use mydb然后向其中插入一条数据db.mydb.insert({_id:1,name:"李小红"})执行完成后，点击测评即可过关哦！第二关：创建集合还是一样的哦，输入 mongo命令，进入到mongodb数据库中首先呢，需要使用 Testdb2数据库use Testdb2然后呢，我们需要向集合中插入题目要求给的数据哦db.t_stu.insert([{_id:1,name:"小明",sex:"男","hobbies":["乒乓球","羽毛球"]},{_id:2,name:"小红",sex:"女","hobbies":["画画","唱歌"]}])这样，我们就已经把数据添加好了哦！第三关：文档操作1输入 mongo命令，进入到mongodb数据库中进入之后，我们我们需要使用 Testdb3数据库use Testdb3然后我们需要document变量document=({_id:1,name:"张小华",sex:"男",phone:"12356986594",hobbies:["打篮球","踢足球","唱歌"]});然后分别插入到stu1、stu2、stu3集合中db.stu1.insert(document); db.stu2.insert(document); db.stu3.insert(document);根据题意，我们需要将stu2的值给修改掉db.stu2.update({phone:"12356986594"},{$set:{phone:"18356971462"}});我们需要通过save方法来将stu3的集合给替换掉，就可以评测了哦db.stu3.save({_id:1,name:"张晓晓",sex:"女",phone:"12365498704",hobbies:["跳舞","羽毛球","唱歌"]});第四关：文档操作二
因为前面三关评论区已经有人把答案发出来了，可以直接参考这些同学步骤来做，就可以通过。（如果有什么问题，可以在底部留言或者QQ上找我都可以哦！）主要是 第四关没有步骤，这个可以跟着我来做哦！第1关：数据的导入导出第二关：高级查询(一)第三关：高级查询(二)第四关：游标首先我们需要切换到 命令行中然后输入 mongo进入到mongodb的命令行中如果出现 > 则证明已经进来了接下来，需要分别执行这三句（截图最后一个是错的，可以根据下面所提供代码进行输入）use mydb4; db.createCollection("test"); for(var i=0;i<10000;i++)db.test.insert({_id:i,title:"MongoDB"+i,content:"hello"+i})然后退出命令（exit）最后执行这句话，点击评测就可以了哦mongoexport -d mydb4 -c test -o /home/test/test4.csv --type csv -f "_id,title,content"
第1关：启动 MongoDB首先，我们进入到命令行中，分别创建以下三个文件夹然后我们需要在 etc目录下面去创建一个放mongodb配置文件的文件夹，并在这个文件夹中创建其配置文件在这个配置文件中添加如下配置，保存后并退出最后启动 MongoDB，点击评测第2关：启动 MongoDB 多实例来到这一关，我们为了以防万一，先重置一下命令行。还是一样的，我们需要创建几个文件夹我们需要移动到配置目录下面创建我们的第一个配置文件在配置文件中添加如下内容port=27021 dbpath=/data/mydb1 logpath=/logs/mymongo1/mongo.log logappend=true fork=true保存退出后，我们需要再拷贝一份，后缀1改为2然后我们需要修改配置文件2修改内容如下然后我们需要分别启动这两个Mongodb的实例，启动完成后点击测评即可。第三关：退出客户端和关闭 MongoDB 服务注意哦：你的端口号可能和我的端口号不是一样的，注意箭头这里的进程号哦
注意，我们首页需要在命令行中启动Hbase(start-hbase.sh)。不要直接复制，最好手敲一遍哦！可以加深一下印象哈。第一关: HBase的MapReduce快速入门package com.processdata; import java.io.IOException; import java.util.List; import java.util.Scanner; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.conf.Configured; import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.client.Connection; import org.apache.hadoop.hbase.client.Put; import org.apache.hadoop.hbase.client.Result; import org.apache.hadoop.hbase.client.Scan; import org.apache.hadoop.hbase.io.ImmutableBytesWritable; import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil; import org.apache.hadoop.hbase.mapreduce.TableMapper; import org.apache.hadoop.hbase.mapreduce.TableReducer; import org.apache.hadoop.hbase.util.Bytes; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.util.Tool; import org.apache.hadoop.util.ToolRunner; import org.apdplat.word.WordSegmenter; import org.apdplat.word.segmentation.Word; import com.util.HBaseUtil; import com.vdurmont.emoji.EmojiParser; /** * 词频统计 * */ public class WorldCountMapReduce extends Configured implements Tool { private static class MyMapper extends TableMapper<Text, IntWritable> { private static byte[] family = "comment_info".getBytes(); private static byte[] column = "content".getBytes(); @Override protected void map(ImmutableBytesWritable rowKey, Result result, Context context) { try { byte[] value = result.getValue(family, column); String content = new String(value, "utf-8"); String[] split = content.split(" "); for (String str : split) { Text text = new Text(str); IntWritable v = new IntWritable(1); context.write(text, v); } } catch (Exception e) { e.printStackTrace(); } } } private static class MyReducer extends TableReducer<Text, IntWritable, ImmutableBytesWritable> { private static byte[] family = "word_info".getBytes(); private static byte[] column = "count".getBytes(); @Override public void reduce(Text key, Iterable<IntWritable> values, Context context) { int sum = 0; for (IntWritable value : values) { sum += value.get(); } Put put = new Put(Bytes.toBytes(key.toString())); put.addColumn(family, column, Bytes.toBytes(sum)); try { context.write(null, put); } catch (IOException e) { // TODO Auto-generated catch block e.printStackTrace(); } catch (InterruptedException e) { // TODO Auto-generated catch block e.printStackTrace(); } } } @Override public int run(String[] args) throws Exception { // 配置Job /********** Begin *********/ // 创建Conf对象 Configuration conf = HBaseConfiguration.create(getConf()); String tablename = args[0]; // 表名 String targetTable = args[1]; // 目标表 // 获取到Job对象 Job job = Job.getInstance(conf); // 创建Scan对象 Scan scan = new Scan(); // 通过Hbase工具类提交数据 TableMapReduceUtil.initTableMapperJob(tablename, scan, MyMapper.class, Text.class, IntWritable.class, job); TableMapReduceUtil.initTableReducerJob(targetTable, MyReducer.class, job); // 开始提交数据 job.waitForCompletion(true); return 0; /********** End *********/ } }第二关：HBase的MapReduce使用package com.processdata; import java.io.IOException; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.conf.Configured; import org.apache.hadoop.fs.Path; import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.client.Put; import org.apache.hadoop.hbase.io.ImmutableBytesWritable; import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil; import org.apache.hadoop.hbase.mapreduce.TableReducer; import org.apache.hadoop.hbase.util.Bytes; import org.apache.hadoop.io.IntWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.lib.input.FileInputFormat; import org.apache.hadoop.mapreduce.lib.input.TextInputFormat; import org.apache.hadoop.util.Tool; /** * 词频统计 * */ public class WorldCountMapReduce2 extends Configured implements Tool { private static class MyMapper extends Mapper<Object, Text, Text, IntWritable> { @Override public void map(Object object, Text value, Context context) throws IOException, InterruptedException { /********** Begin *********/ // 根据题意,我们需要根据空格对指定数据进行拆分 String[] split = value.toString().split(" "); // 循环数组，对值进行分类 for (String str : split) { Text text = new Text(str.getBytes()); IntWritable v = new IntWritable(1); context.write(text, v); } /********** End *********/ } } private static class MyReducer extends TableReducer<Text, IntWritable, ImmutableBytesWritable> { private static byte[] family = "word_info".getBytes(); private static byte[] column = "count".getBytes(); @Override public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException { /********** Begin *********/ int sum = 0; // 用于统计 // 循环Map中分类的值，进行相加 for (IntWritable value : values) { sum += value.get(); } // 将key和value进行聚和 Put put = new Put(Bytes.toBytes(key.toString())); put.addColumn(family, column, Bytes.toBytes(sum)); // 通过文件方式将其输出 context.write(null, put); /********** End *********/ } } @Override public int run(String[] args) throws Exception { // 配置Job /********** Begin *********/ // 配置 Configuration conf = HBaseConfiguration.create(getConf()); String file = args[0]; // 输入文件 String targetTable = args[1]; // 输出表 Job job = Job.getInstance(conf); // Map的Key的输入类型 job.setMapOutputKeyClass(Text.class); // Map的Value的输入类型 job.setMapOutputValueClass(IntWritable.class); // 需要执行的MapReduce类 job.setJarByClass(WorldCountMapReduce2.class); // 文件输入格式 FileInputFormat.addInputPath(job, new Path(file)); // 设置Mapper类 job.setMapperClass(MyMapper.class); // 开始执行任务 TableMapReduceUtil.initTableReducerJob(targetTable, MyReducer.class, job); job.waitForCompletion(true); return 0; /********** End *********/ } }
第一关：批量获取数据package step1; import java.io.IOException; import java.util.ArrayList; import java.util.List; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.Cell; import org.apache.hadoop.hbase.CellScanner; import org.apache.hadoop.hbase.CellUtil; import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.TableName; import org.apache.hadoop.hbase.client.Admin; import org.apache.hadoop.hbase.client.Connection; import org.apache.hadoop.hbase.client.ConnectionFactory; import org.apache.hadoop.hbase.client.Get; import org.apache.hadoop.hbase.client.HTable; 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.client.Table; import org.apache.hadoop.hbase.client.TableDescriptor; import org.apache.hadoop.hbase.generated.rest.rest_jsp; import org.apache.hadoop.hbase.util.Bytes; public class Task { public void batchGet() throws Exception { /********* Begin *********/ Configuration config = HBaseConfiguration.create(); Connection Connection = ConnectionFactory.createConnection(config); List<String> rows = new ArrayList<>(); rows.add("2018"); rows.add("2020"); TableName tableName = TableName.valueOf(Bytes.toBytes("step1_student")); Table table = Connection.getTable(tableName); getData(table,rows); /********* End *********/ } public List<String> getData(Table table, List<String> rows) throws Exception { List<Get> gets = new ArrayList<>(); for (String str : rows) { Get get = new Get(Bytes.toBytes(str)); gets.add(get); } List<String> values = new ArrayList<>(); Result[] results = table.get(gets); for (Result result : results) { System.out.println("Row:" + Bytes.toString(result.getRow())); for (Cell kv : result.rawCells()) { String family = Bytes.toString(CellUtil.cloneFamily(kv)); String qualifire = Bytes.toString(CellUtil.cloneQualifier(kv)); String value = Bytes.toString(CellUtil.cloneValue(kv)); values.add(value); System.out.println(family + ":" + qualifire + "\t" + value); } } return values; } }第二关：批量删除数据package step2; import java.io.IOException; import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; 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.TableDescriptors; import org.apache.hadoop.hbase.TableName; import org.apache.hadoop.hbase.client.*; import org.apache.hadoop.hbase.util.Bytes; public class Task { public void batchDelete()throws Exception{ /********* Begin *********/ Configuration config = HBaseConfiguration.create(); Connection conn = ConnectionFactory.createConnection(config); TableName tableName = TableName.valueOf("step2_table"); Table table = conn.getTable(tableName); //获取表 List<Delete> deletes = new ArrayList<>(); for(int i = 1 ; i < 6;i++){ byte[] row = Bytes.toBytes("row" + i); Delete delete = new Delete(row); deletes.add(delete); } for(int i = 7 ; i < 11;i++){ byte[] row = Bytes.toBytes("row" + i); Delete delete = new Delete(row); deletes.add(delete); } table.delete(deletes); /********* End *********/ } }第三关：批量导入数据至Hbasepackage step3; import java.io.IOException; import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; 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.TableDescriptors; import org.apache.hadoop.hbase.TableName; import org.apache.hadoop.hbase.client.Admin; import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor; import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder; import org.apache.hadoop.hbase.client.Connection; import org.apache.hadoop.hbase.client.ConnectionFactory; import org.apache.hadoop.hbase.client.Get; 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.client.Table; import org.apache.hadoop.hbase.client.TableDescriptor; import org.apache.hadoop.hbase.client.TableDescriptorBuilder; import org.apache.hadoop.hbase.util.Bytes; public class Task { public void batchPut() throws Exception { /********* Begin *********/ Configuration config = new Configuration(); Connection conn = ConnectionFactory.createConnection(config); Admin admin = conn.getAdmin(); TableName tableName = TableName.valueOf(Bytes.toBytes("stu")); TableDescriptorBuilder builder = TableDescriptorBuilder.newBuilder(tableName); ColumnFamilyDescriptor family = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes("basic_info")).build(); ColumnFamilyDescriptor family2 = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes("school_info")).build(); builder.setColumnFamily(family); builder.setColumnFamily(family2); admin.createTable(builder.build()); List<Put> puts = new ArrayList<>(); String[] rows = {"20181122", "20181123"}; String[][] basic_infos = {{"阿克蒙德", "male", "1987-05-23", "tel:139********", "HUNan-ChangSha"}, {"萨格拉斯", "male", "1986-05-23", "tel:187********", "HUNan-ChangSha"}}; String[] basic_colums = {"name", "gender", "birthday", "connect", "address"}; String[][] school_infos = {{"ChengXing", "class 1 grade 2", "Software"}, {"ChengXing", "class 2 grade 2", "Software"}}; String[] school_colums = {"college", "class", "object"}; for (int x = 0; x < rows.length; x++) { // 循环添加数据 Put put = new Put(Bytes.toBytes(rows[x])); for (int i = 0; i < basic_infos.length; i++) { byte[] columFamily = Bytes.toBytes("basic_info"); byte[] qualifier = Bytes.toBytes(basic_colums[i]); byte[] value = Bytes.toBytes(basic_infos[x][i]); put.addColumn(columFamily, qualifier, value); } for (int i = 0; i < school_infos.length; i++) { byte[] columFamily = Bytes.toBytes("school_info"); byte[] qualifier = Bytes.toBytes(school_colums[i]); byte[] value = Bytes.toBytes(school_infos[x][i]); put.addColumn(columFamily, qualifier, value); } puts.add(put); } Table table = conn.getTable(tableName); table.put(puts); /********* End *********/ } }
第一关：JavaApi获取表的列表首先我们进入到命令行先启动hbase然后我们就可以开始写第一关的代码了(注意看前面的标号哦，可能有些会截重复)第二关：修改表第三关：禁用表、启用表、删除表