如何使用PySpark将JSON数据导入HBase：以ticid为行键

阿华AIGC实验室

2026-5-28

使用PySpark将JSON数据导入HBase（以ticid为行键，单列族存储）

没问题，我来一步步带你实现这个需求——用PySpark把你的JSON数据导入HBase，指定ticid作为行键，剩下所有字段都存入单个列族里。

1. 前期准备

确保你的环境已经配置好PySpark和HBase，并且PySpark能访问到HBase的依赖jar包（比如hbase-client、hbase-common），如果提交任务时遇到依赖问题，可以用spark-submit --jars /path/to/hbase-jars/*.jar来指定。
提前在HBase里创建目标表，比如咱们把表名设为ticket_data，列族叫all_details，用HBase Shell执行：

create 'ticket_data', 'all_details'

2. 读取JSON数据

先把你的JSON文件读成Spark DataFrame，不管是本地文件还是HDFS上的都可以：

from pyspark.sql import SparkSession

# 初始化SparkSession
spark = SparkSession.builder \
    .appName("JSON_to_HBase") \
    .getOrCreate()

# 读取JSON文件，本地路径直接写绝对路径，HDFS的话换成hdfs://开头的路径
df = spark.read.json("/path/to/your/ticket_sample.json")

# 可以先看看数据结构确认一下
df.printSchema()

3. 转换数据适配HBase格式

HBase要求每条数据是(rowkey, [(列族:列名, 值), ...])的结构，而且嵌套字段/数组直接存的话不太方便，咱们可以把除了ticid之外的所有字段（包括嵌套的数组、对象）转成JSON字符串，或者拆成扁平化的列——这里先给你两种方案：

方案A：将非行键字段打包成JSON存入列族（简单直接）

这种方式适合不需要单独查询嵌套子字段的场景，把所有非ticid的内容转成一个JSON字符串，或者每个顶层字段转成JSON存在列族的对应列里：

from pyspark.sql.functions import col, to_json

# 获取所有非ticid的字段名
non_key_cols = [col_name for col_name in df.columns if col_name != "ticid"]

# 把每个顶层字段转成JSON字符串，列名格式为`all_details:字段名`
transformed_df = df.select(
    col("ticid").alias("rowkey"),
    *[to_json(col(c)).alias(f"all_details:{c}") for c in non_key_cols]
)

# 转换成HBase需要的RDD格式：(rowkey字节数组, [(列名字节数组, 值字节数组), ...])
hbase_rdd = transformed_df.rdd.map(lambda row: (
    row.rowkey.encode('utf-8'),
    [(k.encode('utf-8'), v.encode('utf-8')) for k, v in row.asDict().items() if k != "rowkey"]
))

方案B：扁平化嵌套字段（适合需要查询子字段的场景）

如果之后需要单独查询commentno、Traino这类子字段，咱们可以把嵌套结构扁平化，处理数组的时候可以用explode（注意这会把数组的每个元素拆成单独的行，rowkey可能重复，需要给rowkey加后缀区分）：

from pyspark.sql.functions import explode, col

# 先处理Comments里的数组
comment_df = df.select(
    col("ticid"),
    col("ticlocation"),
    col("custnum"),
    explode(col("Comments.comment")).alias("comment"),
    col("Rails")
)

# 再展开comment里的字段
flat_comment_df = comment_df.select(
    col("ticid"),
    col("ticlocation"),
    col("custnum"),
    col("comment.commentno"),
    col("comment.desc"),
    col("comment.passengerseat.intele").alias("passengerseat_intele"),
    col("comment.passengerloc.intele").alias("passengerloc_intele"),
    col("Rails")
)

# 同理处理Rails数组（这里省略，你可以按同样的方式展开）
# 最后转换为HBase格式，每个扁平化后的字段对应列族的一个列

4. 写入HBase

这里给你两种写入方式，按需选择：

方式一：用HappyBase客户端（适合小规模数据）

先安装HappyBase：pip install happybase，然后批量写入：

import happybase

# 连接HBase的ZooKeeper
conn = happybase.Connection('your_zookeeper_host', port=2181)
table = conn.table('ticket_data')

# 批量写入，batch_size可以根据数据量调整
with table.batch(batch_size=1000) as batch:
    for rowkey, columns in hbase_rdd.collect():
        batch.put(rowkey, dict(columns))

conn.close()

方式二：用Spark的Hadoop API（适合大规模数据，推荐）

这种方式更适合大数据量的场景，不需要把数据拉到本地：

# 配置HBase参数，替换成你的ZooKeeper地址
hbase_conf = {
    "hbase.zookeeper.quorum": "your_zookeeper_host",
    "hbase.mapreduce.outputtable": "ticket_data",
    "mapreduce.outputformat.class": "org.apache.hadoop.hbase.mapreduce.TableOutputFormat",
    "mapreduce.job.output.key.class": "org.apache.hadoop.hbase.io.ImmutableBytesWritable",
    "mapreduce.job.output.value.class": "org.apache.hadoop.hbase.client.Put"
}

# 导入HBase的Java类
from org.apache.hadoop.hbase.io import ImmutableBytesWritable
from org.apache.hadoop.hbase.client import Put
from org.apache.hadoop.hbase.util import Bytes

# 定义转换函数，把RDD的每一行转换成HBase的Put对象
def create_hbase_put(row):
    rowkey, columns = row
    put = Put(Bytes.toBytes(rowkey))
    for col, val in columns:
        # 拆分列族和列名
        family, qualifier = col.split(b':', 1)
        put.addColumn(family, qualifier, val)
    return (ImmutableBytesWritable(), put)

# 转换RDD格式
hbase_put_rdd = hbase_rdd.map(create_hbase_put)

# 写入HBase
hbase_put_rdd.saveAsNewAPIHadoopDataset(conf=hbase_conf)