当前位置： 首页 > wzjs >正文

有没有做高仿手表的网站网络运营是做什么的工作

wzjs 2025/8/11 5:08:28
有没有做高仿手表的网站,网络运营是做什么的工作,快速搭建网站教程,网站访问很慢Spark读写HBase表 摘要一、实验环境准备1. 技术版本2. Maven 依赖配置 二、实验步骤1. 数据准备2. HBase 表结构设计3. 代码实现3.1 数据写入 HBase（writeDataToHBase 方法）3.2 数据读取与分析（readHBaseData 方法3.3 Spark SQL 分析3.4 完整…

Spark读写HBase表

  • 摘要
  • 一、实验环境准备
    • 1. 技术版本
    • 2. Maven 依赖配置
  • 二、实验步骤
    • 1. 数据准备
    • 2. HBase 表结构设计
    • 3. 代码实现
      • 3.1 数据写入 HBase（`writeDataToHBase` 方法）
      • 3.2 数据读取与分析（`readHBaseData` 方法
      • 3.3 Spark SQL 分析
      • 3.4 完整代码
  • 三、实验结果
    • 1. 数据写入验证
    • 2. 数据读取与分析结果
  • 四、常见问题与优化
    • 1. 空值处理
    • 2. 性能优化
    • 3. 集群配置（可选做）
  • 五、总结

摘要

本文介绍如何使用 Spark 2.3.2 实现对 HBase 1.4.8 表的读写操作，通过 Scala 语言将 CSV 数据写入 HBase，并利用 Spark SQL 分析数据。代码示例涵盖数据批量写入、全表扫描、数据类型转换及结构化查询，适合大数据开发人员快速掌握 Spark 与 HBase 的集成方法。

一、实验环境准备

1. 技术版本

  • Spark：2.3.2
  • HBase：1.4.8
  • Scala：2.11
  • 开发工具：IntelliJ IDEA
  • 依赖管理：Maven

2. Maven 依赖配置

在 pom.xml 中添加以下依赖：

<dependencies>  <!-- Spark 核心与 SQL -->  <dependency>  <groupId>org.apache.spark</groupId>  <artifactId>spark-core_2.11</artifactId>  <version>2.3.2</version>  </dependency>  <dependency>  <groupId>org.apache.spark</groupId>  <artifactId>spark-sql_2.11</artifactId>  <version>2.3.2</version>  </dependency>  <!-- HBase 客户端与 MapReduce 支持 -->  <dependency>  <groupId>org.apache.hbase</groupId>  <artifactId>hbase-client</artifactId>  <version>1.4.8</version>  </dependency>  <dependency>  <groupId>org.apache.hbase</groupId>  <artifactId>hbase-common</artifactId>  <version>1.4.8</version>  </dependency>  <dependency>  <groupId>org.apache.hbase</groupId>  <artifactId>hbase-server</artifactId>  <version>1.4.8</version>  </dependency>  <dependency>  <groupId>org.apache.hbase</groupId>  <artifactId>hbase-mapreduce</artifactId>  <version>1.4.8</version>  </dependency>  <!-- Hadoop 客户端（与 HBase 兼容） -->  <dependency>  <groupId>org.apache.hadoop</groupId>  <artifactId>hadoop-client</artifactId>  <version>2.7.3</version>  </dependency>  
</dependencies>

二、实验步骤

1. 数据准备

  • 文件路径：D:\\JavaProjects\\SparkAllProjects\\data\\emp.csv
  • 数据格式（示例）：
    7369,SMITH,CLERK,7902,1980-12-17,800,,20  
7499,ALLEN,SALESMAN,7698,1981-02-20,1600,300,30
    字段含义：员工ID,姓名,职位,上级ID,入职日期,薪资,奖金,部门ID

2. HBase 表结构设计

  • 表名：employee
  • 列族：info
  • 列标识：
    • ename：姓名
    • job：职位
    • mgr：上级ID
    • hiredate：入职日期
    • salary：薪资
    • comm：奖金
    • deptNo：部门ID

3. 代码实现

3.1 数据写入 HBase（writeDataToHBase 方法）

  • 核心逻辑：
    1. 读取 CSV 文件为 DataFrame。
    2. 按分区遍历数据，批量创建 Put 对象。
    3. 通过 HBase 连接将数据写入表中，避免单条写入性能瓶颈。
  • 核心代码
      def writeDataToHBase(spark: SparkSession): Unit = {// 2.读取数据文件val empDF = spark.read.csv("file:///D:\\JavaProjects\\SparkAllProjects\\data\\emp.csv")// 3.按照DataFrame分区写入HBase表中empDF.foreachPartition(p => {// 3.1 配置HBase连接地址：初始化conf配置对象、配置zk连接地址及其端口val conf = HBaseConfiguration.create()conf.set("hbase.zookeeper.quorum", "s1,s2,s3")conf.set("hbase.zookeeper.property.clientPort", "2181")// 3.2 在每个HBase节点中创建HBase的链接对象val conn = ConnectionFactory.createConnection(conf)// 3.3 获取HBase目标表val table = conn.getTable(TableName.valueOf(TABLE_NAME))// 批量提交时，使用list列表存储put，当达到batchSize大小时提交一次val batchSize = 14 // emp.csv就14条记录，所以设置成14条提交一次var puts = List[Put]() // puts数组// 3.4 将dataframe中的每个分区数据写入table表中try {p.foreach(row => {// 3.5 获取每行中的各个列的数据val empNo = row.getString(0)val ename = row.getString(1)val job = row.getString(2)val mgr = row.getString(3)val hireDate = row.getString(4)val salary = row.getString(5)val comm = row.getString(6)val deptNo = row.getString(7)var mgrStr = "0"if (mgr != null) {mgrStr = mgr}var commStr = "0.0"if (comm != null) {commStr = comm}// 3.6 设置rowkey：按照empNoval rowKey = Bytes.toBytes(empNo)// 3.7 创建Put对象，设置列族中的列和字段val put = new Put(rowKey)put.addColumn(Bytes.toBytes(CF_NAME), Bytes.toBytes("ename"), Bytes.toBytes(ename))put.addColumn(Bytes.toBytes(CF_NAME), Bytes.toBytes("job"), Bytes.toBytes(job))put.addColumn(Bytes.toBytes(CF_NAME), Bytes.toBytes("mgr"), Bytes.toBytes(mgrStr))put.addColumn(Bytes.toBytes(CF_NAME), Bytes.toBytes("hiredate"), Bytes.toBytes(hireDate))put.addColumn(Bytes.toBytes(CF_NAME), Bytes.toBytes("salary"), Bytes.toBytes(salary))put.addColumn(Bytes.toBytes(CF_NAME), Bytes.toBytes("comm"), Bytes.toBytes(commStr))put.addColumn(Bytes.toBytes(CF_NAME), Bytes.toBytes("deptNo"), Bytes.toBytes(deptNo))// 3.8 将该Put对象加入Table对象中puts = put :: putsif (puts.size >= batchSize) {// 转成Java中的ArrayListval javaLists = new util.ArrayList[Put](puts.size)puts.foreach(javaLists.add)// table调用put添加javaListstable.put(javaLists)// 添加完成后，清空puts = List[Put]()}// 处理剩余数据: 当不满足14条数据时，把剩余数据写入HBase表中if (puts.nonEmpty) {val javaLists = new util.ArrayList[Put](puts.size)puts.foreach(javaLists.add)table.put(javaLists)}})} finally {// 确保资源释放if (table != null) table.close()if (conn != null) conn.close()}})}

3.2 数据读取与分析（readHBaseData 方法

  • 核心逻辑：
    1. 使用 ResultScanner 全表扫描 HBase 数据。
    2. 将二进制数据转换为样例类 Employee，自动推断 DataFrame 的 Schema。
    3. 通过 Spark SQL 执行聚合查询（如按部门统计薪资总和）。
  • 核心代码：
    case class Employee(  employee_id: Int,  employee_name: String,  job_title: String,  manager_id: Int,  hire_date: String,  salary: Double,  bonus: Double,  department_id: Int  
)  def readHBaseData(spark: SparkSession): DataFrame = {// 1. 配置 HBase 连接参数val conf = HBaseConfiguration.create()conf.set("hbase.zookeeper.quorum", "s1,s2,s3")  // 替换为你的 ZK 地址conf.set("hbase.zookeeper.property.clientPort", "2181")// 2. 创建 HBase 连接和表对象val conn = ConnectionFactory.createConnection(conf)val table = conn.getTable(TableName.valueOf(TABLE_NAME))var scanner: ResultScanner = nulltry {// 3. 构造扫描器（Scan）并配置val scan = new Scan().addFamily(Bytes.toBytes(CF_NAME))  // 读取指定列族下的所有列.setCaching(500)  // 提升批量读取性能.setCacheBlocks(false)// 4. 获取扫描结果迭代器scanner = table.getScanner(scan)// 5. 遍历结果并转换为 Employee 对象val employees = ListBuffer[Employee]()val it = scanner.iterator()while (it.hasNext) {val result: Result = it.next()// 提取行键（假设 rowkey 是 employee_id 的字符串形式）val rowKeyStr = Bytes.toString(result.getRow)val employeeId = rowKeyStr.toInt  // 转换为 Int（需确保 rowkey 是数字）// 提取各列数据（根据 HBase 实际存储的列名调整）val name = Bytes.toString(result.getValue(Bytes.toBytes(CF_NAME), Bytes.toBytes("ename")))val job = Bytes.toString(result.getValue(Bytes.toBytes(CF_NAME), Bytes.toBytes("job")))val mgrStr = Bytes.toString(result.getValue(Bytes.toBytes(CF_NAME), Bytes.toBytes("mgr")))val hireDate = Bytes.toString(result.getValue(Bytes.toBytes(CF_NAME), Bytes.toBytes("hiredate")))val salaryStr = Bytes.toString(result.getValue(Bytes.toBytes(CF_NAME), Bytes.toBytes("salary")))val commStr = Bytes.toString(result.getValue(Bytes.toBytes(CF_NAME), Bytes.toBytes("comm")))val deptNoStr = Bytes.toString(result.getValue(Bytes.toBytes(CF_NAME), Bytes.toBytes("deptNo")))// 处理可能的空值或转换异常（示例：默认值为 0）val managerId = if (mgrStr != null && mgrStr.nonEmpty) mgrStr.toInt else 0val salary = if (salaryStr != null && salaryStr.nonEmpty) salaryStr.toDouble else 0val bonus = if (commStr != null && commStr.nonEmpty) commStr.toDouble else 0val departmentId = if (deptNoStr != null && deptNoStr.nonEmpty) deptNoStr.toInt else 0// 创建 Employee 对象并添加到列表employees += Employee(employee_id = employeeId,employee_name = name,job_title = job,manager_id = managerId,hire_date = hireDate,salary = salary,bonus = bonus,department_id = departmentId)}// 6. 将 Employee 列表转换为 DataFrame（自动推断 Schema）import spark.implicits._spark.createDataFrame(employees.toList)} catch {case e: Exception =>println(s"读取 HBase 数据失败: ${e.getMessage}")throw e  // 抛异常终止流程} finally {// 7. 释放所有资源（关键！避免连接泄漏）if (scanner != null) scanner.close()if (table != null) table.close()if (conn != null) conn.close()}}

3.3 Spark SQL 分析

  • 读取数据后，通过 Spark SQL 执行聚合查询：
    val df = readHBaseData(spark)  
df.createOrReplaceTempView("emp")  
spark.sql("""  SELECT department_id,  SUM(salary + bonus) AS total  FROM emp  GROUP BY department_id  ORDER BY total DESC  
""").show(false)
  • 其他分析查询请自行操作。

3.4 完整代码

  • 在IDEA中创建名为WriteAndReadDataToHBase单例对象
  • 添加如下完成代码：
    package com.lpssfxy.spark.datasourceimport org.apache.hadoop.hbase.{HBaseConfiguration, TableName}
import org.apache.hadoop.hbase.client.{ConnectionFactory, Put, Result, ResultScanner, Scan}
import org.apache.hadoop.hbase.util.Bytes
import org.apache.spark.sql.{DataFrame, SparkSession}import java.util
import scala.collection.mutable.ListBuffercase class Employee(employee_id: Int,       // 对应 HBase rowkey（假设 rowkey 是员工ID）employee_name: String,  // 对应列族 info:enamejob_title: String,      // 对应列族 info:jobmanager_id: Int,        // 对应列族 info:mgrhire_date: String,      // 对应列族 info:hiredatesalary: Double,            // 对应列族 info:salarybonus: Double,             // 对应列族 info:commdepartment_id: Int      // 对应列族 info:deptNo)/*** spark读写HBase表employee*/
object WriteAndReadDataToHBase {// 定义表名称及其列族名称private val TABLE_NAME = "employee"private val CF_NAME = "info"def main(args: Array[String]): Unit = {// 1. 准备环境：SparkSession初始化val spark = SparkSession.builder().appName("WriteDataToHBase").master("local[*]").getOrCreate()// 2.调用writeDataToHBase方法将文件写入HBase中//writeDataToHBase(spark)// 3. 读HBase数据val df = readHBaseData(spark)//df.show(false)df.createOrReplaceTempView("emp")spark.sql("select department_id,sum(salary+bonus) as total from emp group by department_id order by total desc").show(false)// 4.停止SparkSession对象，释放资源spark.stop()}/*** 写数据到HBase表中* @param spark*/def writeDataToHBase(spark: SparkSession): Unit = {// 2.读取数据文件val empDF = spark.read.csv("file:///D:\\JavaProjects\\SparkAllProjects\\data\\emp.csv")// 3.按照DataFrame分区写入HBase表中empDF.foreachPartition(p => {// 3.1 配置HBase连接地址：初始化conf配置对象、配置zk连接地址及其端口val conf = HBaseConfiguration.create()conf.set("hbase.zookeeper.quorum", "s1,s2,s3")conf.set("hbase.zookeeper.property.clientPort", "2181")// 3.2 在每个HBase节点中创建HBase的链接对象val conn = ConnectionFactory.createConnection(conf)// 3.3 获取HBase目标表val table = conn.getTable(TableName.valueOf(TABLE_NAME))// 批量提交时，使用list列表存储put，当达到batchSize大小时提交一次val batchSize = 14 // emp.csv就14条记录，所以设置成14条提交一次var puts = List[Put]() // puts数组// 3.4 将dataframe中的每个分区数据写入table表中try {p.foreach(row => {// 3.5 获取每行中的各个列的数据val empNo = row.getString(0)val ename = row.getString(1)val job = row.getString(2)val mgr = row.getString(3)val hireDate = row.getString(4)val salary = row.getString(5)val comm = row.getString(6)val deptNo = row.getString(7)var mgrStr = "0"if (mgr != null) {mgrStr = mgr}var commStr = "0.0"if (comm != null) {commStr = comm}// 3.6 设置rowkey：按照empNoval rowKey = Bytes.toBytes(empNo)// 3.7 创建Put对象，设置列族中的列和字段val put = new Put(rowKey)put.addColumn(Bytes.toBytes(CF_NAME), Bytes.toBytes("ename"), Bytes.toBytes(ename))put.addColumn(Bytes.toBytes(CF_NAME), Bytes.toBytes("job"), Bytes.toBytes(job))put.addColumn(Bytes.toBytes(CF_NAME), Bytes.toBytes("mgr"), Bytes.toBytes(mgrStr))put.addColumn(Bytes.toBytes(CF_NAME), Bytes.toBytes("hiredate"), Bytes.toBytes(hireDate))put.addColumn(Bytes.toBytes(CF_NAME), Bytes.toBytes("salary"), Bytes.toBytes(salary))put.addColumn(Bytes.toBytes(CF_NAME), Bytes.toBytes("comm"), Bytes.toBytes(commStr))put.addColumn(Bytes.toBytes(CF_NAME), Bytes.toBytes("deptNo"), Bytes.toBytes(deptNo))// 3.8 将该Put对象加入Table对象中puts = put :: putsif (puts.size >= batchSize) {// 转成Java中的ArrayListval javaLists = new util.ArrayList[Put](puts.size)puts.foreach(javaLists.add)// table调用put添加javaListstable.put(javaLists)// 添加完成后，清空puts = List[Put]()}// 处理剩余数据: 当不满足14条数据时，把剩余数据写入HBase表中if (puts.nonEmpty) {val javaLists = new util.ArrayList[Put](puts.size)puts.foreach(javaLists.add)table.put(javaLists)}})} finally {// 确保资源释放if (table != null) table.close()if (conn != null) conn.close()}})}/*** 读取HBase表数据* @param spark* @return*/def readHBaseData(spark: SparkSession): DataFrame = {// 1. 配置 HBase 连接参数val conf = HBaseConfiguration.create()conf.set("hbase.zookeeper.quorum", "s1,s2,s3")  // 替换为你的 ZK 地址conf.set("hbase.zookeeper.property.clientPort", "2181")// 2. 创建 HBase 连接和表对象val conn = ConnectionFactory.createConnection(conf)val table = conn.getTable(TableName.valueOf(TABLE_NAME))var scanner: ResultScanner = nulltry {// 3. 构造扫描器（Scan）并配置val scan = new Scan().addFamily(Bytes.toBytes(CF_NAME))  // 读取指定列族下的所有列.setCaching(500)  // 提升批量读取性能.setCacheBlocks(false)// 4. 获取扫描结果迭代器scanner = table.getScanner(scan)// 5. 遍历结果并转换为 Employee 对象val employees = ListBuffer[Employee]()val it = scanner.iterator()while (it.hasNext) {val result: Result = it.next()// 提取行键（假设 rowkey 是 employee_id 的字符串形式）val rowKeyStr = Bytes.toString(result.getRow)val employeeId = rowKeyStr.toInt  // 转换为 Int（需确保 rowkey 是数字）// 提取各列数据（根据 HBase 实际存储的列名调整）val name = Bytes.toString(result.getValue(Bytes.toBytes(CF_NAME), Bytes.toBytes("ename")))val job = Bytes.toString(result.getValue(Bytes.toBytes(CF_NAME), Bytes.toBytes("job")))val mgrStr = Bytes.toString(result.getValue(Bytes.toBytes(CF_NAME), Bytes.toBytes("mgr")))val hireDate = Bytes.toString(result.getValue(Bytes.toBytes(CF_NAME), Bytes.toBytes("hiredate")))val salaryStr = Bytes.toString(result.getValue(Bytes.toBytes(CF_NAME), Bytes.toBytes("salary")))val commStr = Bytes.toString(result.getValue(Bytes.toBytes(CF_NAME), Bytes.toBytes("comm")))val deptNoStr = Bytes.toString(result.getValue(Bytes.toBytes(CF_NAME), Bytes.toBytes("deptNo")))// 处理可能的空值或转换异常（示例：默认值为 0）val managerId = if (mgrStr != null && mgrStr.nonEmpty) mgrStr.toInt else 0val salary = if (salaryStr != null && salaryStr.nonEmpty) salaryStr.toDouble else 0val bonus = if (commStr != null && commStr.nonEmpty) commStr.toDouble else 0val departmentId = if (deptNoStr != null && deptNoStr.nonEmpty) deptNoStr.toInt else 0// 创建 Employee 对象并添加到列表employees += Employee(employee_id = employeeId,employee_name = name,job_title = job,manager_id = managerId,hire_date = hireDate,salary = salary,bonus = bonus,department_id = departmentId)}// 6. 将 Employee 列表转换为 DataFrame（自动推断 Schema）import spark.implicits._spark.createDataFrame(employees.toList)} catch {case e: Exception =>println(s"读取 HBase 数据失败: ${e.getMessage}")throw e  // 抛异常终止流程} finally {// 7. 释放所有资源（关键！避免连接泄漏）if (scanner != null) scanner.close()if (table != null) table.close()if (conn != null) conn.close()}}}

三、实验结果

1. 数据写入验证

  • 控制台输出无异常日志，HBase 表 employee 中生成对应 rowkey 的记录。
  • 通过 HBase Shell 命令 scan 'employee' 可查看数据：
    hbase> scan 'employee', {LIMIT => 2}  
ROW       COLUMN+CELL                                              
7369      column=info:ename, timestamp=... value=SMITH               
7369      column=info:job, timestamp=... value=CLERK               
# ...（其他列省略）

2. 数据读取与分析结果

  • 原始数据展示：

    +-----------+-------------+-----------+-----------+----------+------+-----+-------------+  
|employee_id|employee_name|job_title  |manager_id |hire_date |salary|bonus|department_id|  
+-----------+-------------+-----------+-----------+----------+------+-----+-------------+  
|7369       |SMITH        |CLERK      |7902       |1980-12-17|800.0 |0.0  |20           |  
|7499       |ALLEN        |SALESMAN   |7698       |1981-02-20|1600.0|300.0|30           |  
# ...（其他行省略）

  • Spark SQL 聚合结果：

    +-------------+------------+  
|department_id|total       |  
+-------------+------------+  
|30           |20150.0     |  
|20           |10875.0     |  
|10           |8750.0      |  
+-------------+------------+

四、常见问题与优化

1. 空值处理

  • HBase 列值为 null 时，result.getValue 返回 null，需通过 Option 或判空逻辑处理：
    val commStr = Option(result.getValue(...)).map(Bytes.toString).getOrElse("0.0")

2. 性能优化

  • 批量写入：调整 batchSize（建议 500-2000），减少 RPC 调用次数。
  • 扫描缓存：通过 scan.setCaching(500) 提升全表扫描效率。
  • 数据类型：避免过度使用字符串类型，对数值字段直接存储二进制数据（如 Bytes.toDouble）。

3. 集群配置（可选做）

  • 在 spark-submit 中添加 HBase 配置文件（如 hbase-site.xml），确保 Executor 节点访问 ZooKeeper：
    spark-submit --files hbase-site.xml ...

五、总结

本文通过实际案例演示了 Spark 与 HBase 的集成流程，实现了从 CSV 数据写入 HBase 到结构化数据分析的完整链路。核心要点包括：

  1. 使用 foreachPartition 实现批量写入，避免单条操作性能损耗。
  2. 通过 ResultScanner 和样例类映射，简化 HBase 数据到 DataFrame 的转换。
  3. 利用 Spark SQL 对 HBase 数据进行高效分析，发挥分布式计算优势。

该实验适用于海量结构化数据的存储与分析场景，可进一步扩展至实时数据处理或机器学习模型训练。

http://www.dtcms.com/wzjs/300807.html

相关文章：

  • 安徽疫情seo网站平台
  • 制作网站协议书友点企业网站管理系统
  • 如何在网站找做贸易的客户武汉网站运营专业乐云seo
  • 建设国家标准官方网站sem竞价推广托管
  • 湛江手机网站建设网站seo推广哪家值得信赖
  • 跨境电商综合服务平台有哪些重庆seo优化
  • 有了空间和域名 网站容易做吗品牌策略
  • 南京网站制作哪家好链接是什么意思
  • 网站单页面可以做302跳转吗友情链接搜读
  • 微信优惠券网站怎么做的南宁seo标准
  • 网站获取qq号码 原理管理方面的培训课程
  • 增加wordpress页脚小工具网站优化 福州
  • 做系统之前的网站襄阳seo培训
  • 盘锦威旺做网站软文推广广告
  • 有个专门做装修的网站北京网站定制公司
  • 网络建站一般多少钱免费seo网站的工具
  • 网站有哪几种seo课程培训
  • 高端网站建站公司seo公司发展前景
  • 郴州网站陕西seo优化
  • 顺德顺的网站建设竞价托管推广
  • 做企业网站的流程潍坊网站定制模板建站
  • 邯郸市住房和城乡建设局官网站长工具seo排名查询
  • 做视频网站赚钱吗网站优化公司上海
  • 网站怎么经营南京网站制作
  • 网站建设总体说明书福州短视频seo
  • 个人备案 做政府网站网站设计师
  • 岳阳网站开发收费北京优化推广
  • 宁乡网站建设uuv9新媒体营销推广公司
  • 建设b2b网站要求百度网站是什么
  • 东莞一站式网站推广运营seo外包网络公司