微风小屋 - 大数据
type: channel
link: https://www.luocome.cn/category/bigdata/
items: Hadoop 伪分布式安装教程（基于Centos7系统）
items: educoder-Flink SQL基础
items: educoder-Spark机器学习
items: educoder-招聘数据清洗
items: educoder-MongoDB 数据库基本操作
items: educoder-MongoDB 文档的高级查询操作
items: educoder-初识 MongoDB
items: educoder-当HBase遇上MapReduce
items: educoder-Hbase开发批量操作
items: educoder-JavaApi管理表

Hadoop 伪分布式安装教程（基于Centos7系统）
type: item
date: 2023-10-26T11:04:00+08:00
description: 先启动虚拟机，并通过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安装成功了

educoder-Flink SQL基础
type: item
date: 2021-06-07T17:18:00+08:00
description: 第一关：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() { r…

educoder-Spark机器学习
type: item
date: 2021-05-31T10:54:00+08:00
description: 感谢宋同学提供的下面内容基于物品推荐算法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表示：其他用户购买的商品与用户…

educoder-招聘数据清洗
type: item
date: 2021-05-08T10:40:00+08:00
description: 注意，我们需要编写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 sq…

educoder-MongoDB 数据库基本操作
type: item
date: 2021-04-26T10:08:00+08:00
description: 第一关：创建数据库首先我们需要来到 命令行中，输入 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:["跳舞","羽毛球","唱歌"]});第四关：文档操作二

educoder-MongoDB 文档的高级查询操作
type: item
date: 2021-04-25T10:56:00+08:00
description: 因为前面三关评论区已经有人把答案发出来了，可以直接参考这些同学步骤来做，就可以通过。（如果有什么问题，可以在底部留言或者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"

educoder-初识 MongoDB
type: item
date: 2021-04-24T18:00:00+08:00
description: 第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 服务注意哦：你的端口号可能和我的端口号不是一样的，注意箭头这里的进程号哦

educoder-当HBase遇上MapReduce
type: item
date: 2021-04-19T16:13:00+08:00
description: 注意，我们首页需要在命令行中启动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 …

educoder-Hbase开发批量操作
type: item
date: 2021-04-18T12:13:00+08:00
description: 第一关：批量获取数据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);…

educoder-JavaApi管理表
type: item
date: 2021-04-08T00:26:00+08:00
description: 第一关：JavaApi获取表的列表首先我们进入到命令行先启动hbase然后我们就可以开始写第一关的代码了(注意看前面的标号哦，可能有些会截重复)第二关：修改表第三关：禁用表、启用表、删除表