本文介紹使用Python的多線程技術(shù)，提高h(yuǎn)appybase模塊和gdal模塊的效率，從tif格式的影像文件中讀取數(shù)據(jù)，并將其存儲(chǔ)到HBase數(shù)據(jù)庫(kù)中。主要步驟包括：

• 準(zhǔn)備工作：安裝Python環(huán)境，安裝happybase模塊和gdal模塊，安裝HBase數(shù)據(jù)庫(kù)，并準(zhǔn)備tif影像文件。
• 讀取tif影像數(shù)據(jù)：使用readTif函數(shù)讀取tif影像數(shù)據(jù)集，并獲取其寬度、高度、波段數(shù)、數(shù)據(jù)數(shù)組、仿射變換參數(shù)和投影信息。遍歷tif影像文件所在的文件夾，獲取tif影像文件的日期和分塊信息。
• 寫入HBase數(shù)據(jù)庫(kù)：創(chuàng)建一個(gè)happybase連接對(duì)象，并獲取或創(chuàng)建一個(gè)happybase表對(duì)象。遍歷每個(gè)分塊，使用readTif函數(shù)讀取每個(gè)分塊的每個(gè)日期的每個(gè)波段的數(shù)據(jù)，并將其存儲(chǔ)到一個(gè)三維的numpy數(shù)組中。遍歷每個(gè)像素，將其對(duì)應(yīng)的每個(gè)波段的每個(gè)日期的數(shù)據(jù)組合成一個(gè)字典，作為HBase表中的列值。使用分塊編號(hào)、行號(hào)和列號(hào)拼接成一個(gè)字符串，作為HBase表中的行鍵。使用put方法將行鍵和列值寫入HBase表中。關(guān)閉happybase連接對(duì)象。
• 使用多線程技術(shù)：導(dǎo)入threading模塊，創(chuàng)建一個(gè)信號(hào)量對(duì)象，用于限制線程的最大數(shù)量。遍歷每個(gè)分塊，使用信號(hào)量對(duì)象的acquire方法獲取一個(gè)信號(hào)量，然后創(chuàng)建一個(gè)新的線程對(duì)象，并指定目標(biāo)函數(shù)為load函數(shù)，以及傳遞分塊編號(hào)、分塊列表和日期列表作為參數(shù)。然后調(diào)用start方法啟動(dòng)該線程。在load函數(shù)中，在完成數(shù)據(jù)的讀取和寫入后，使用信號(hào)量對(duì)象的release方法釋放一個(gè)信號(hào)量，并關(guān)閉happybase連接對(duì)象。記錄程序運(yùn)行的開(kāi)始時(shí)間和結(jié)束時(shí)間，并計(jì)算程序運(yùn)行的總時(shí)間。

一、環(huán)境準(zhǔn)備

• 安裝Python環(huán)境，本文使用的是Anaconda3。
• 安裝happybase模塊，可以使用pip或conda命令。例如：

pip install happybase

• 安裝gdal模塊，可以使用pip或conda命令。例如：

conda install gdal

• 啟動(dòng)分布式集群，hadoop以及hbase；啟動(dòng)thrift服務(wù)，可以使用hbase-daemon.sh腳本。例如：

hbase-daemon.sh start thrift

就可以使用happybase模塊連接到thrift服務(wù)，并操作HBase數(shù)據(jù)庫(kù)了?

• 準(zhǔn)備tif格式的影像文件，并放在一個(gè)文件夾中。本文使用的是Sentinel-2衛(wèi)星的10個(gè)波段的影像數(shù)據(jù)，分為多個(gè)日期和多個(gè)分塊。

二、定義讀取tif影像數(shù)據(jù)的函數(shù)

• 導(dǎo)入需要的模塊，包括time、happybase、gdal、numpy、pandas、os、tqdm和threading。例如：

import time
import happybase
from osgeo import gdal
import numpy as np
import pandas as pd
import os
from tqdm import tqdm
import threading

• 定義一個(gè)函數(shù)readTif，用于讀取tif格式的影像數(shù)據(jù)集，并返回其寬度、高度、波段數(shù)、數(shù)據(jù)數(shù)組、仿射變換參數(shù)和投影信息。例如：

#  讀取tif數(shù)據(jù)集
def readTif(fileName, xoff=0, yoff=0, data_width=0, data_height=0):
    dataset = gdal.Open(fileName)
    num_bands = dataset.RasterCount
    # print(num_bands)
    if dataset == None:
        print(fileName + "文件無(wú)法打開(kāi)")
    #  柵格矩陣的列數(shù)
    width = dataset.RasterXSize
    #  柵格矩陣的行數(shù)
    height = dataset.RasterYSize
    #  波段數(shù)
    bands = dataset.RasterCount
    #  獲取數(shù)據(jù)
    if (data_width == 0 and data_height == 0):
        data_width = width
        data_height = height
    data = dataset.ReadAsArray(xoff, yoff, data_width, data_height)
    #  獲取仿射矩陣信息
    geotrans = dataset.GetGeoTransform()
    #  獲取投影信息
    proj = dataset.GetProjection()
    return width, height, bands, data, geotrans, proj

• ?獲取tif影像文件所在的文件夾路徑，并遍歷該文件夾下所有以.tif為后綴名的文件。例如：

# 分塊影像所在文件夾，不能有中文
tifDir = r"E:\pyimg\tif2csv\S2SR10mallband3tile"
tifs = [i for i in os.listdir(tifDir) if i.endswith(".tif")]
print("有 %s 個(gè)tif文件" % len(tifs))

• ?獲取tif影像文件的日期和分塊信息，并去重排序。例如：

# 獲取目標(biāo)文件數(shù)量，前綴相同的
bandlist=['B2','B3','B4','B5','B6','B7','B8','B8A','B11','B12']
datelist1 = []
fenkuailist1 = []
for i in tifs:
    datelist1.append(i[:-26])
    fenkuailist1.append(i[-25:-4])

datelist = list(set(datelist1))
datelist.sort(key=datelist1.index)
fenkuailist = list(set(fenkuailist1))
fenkuailist.sort(key=fenkuailist1.index)
print("有 %s 個(gè)日期" % len(datelist))
print("datelist" , datelist)
print("每個(gè)日期 %s 個(gè)塊" % len(fenkuailist))
print("fenkuailist" , fenkuailist)

三、創(chuàng)建happybase連接和表對(duì)象

• 創(chuàng)建一個(gè)happybase連接對(duì)象，并指定HBase數(shù)據(jù)庫(kù)的IP地址。例如：

connection = happybase.Connection('192.168.1.100')
# # before first use:
connection.open()

• 獲取或創(chuàng)建一個(gè)happybase表對(duì)象，并指定表名和列族名。例如：

table = connection.table('rawdata')

四、定義寫入HBase數(shù)據(jù)庫(kù)的函數(shù)

• 定義一個(gè)函數(shù)load，用于讀取和寫入一個(gè)分塊的數(shù)據(jù)。該函數(shù)接受分塊編號(hào)、分塊列表和日期列表作為參數(shù)。該函數(shù)的主要步驟如下：文章來(lái)源地址http://www.zghlxwxcb.cn/news/detail-494494.html

  • 打印當(dāng)前分塊的編號(hào)。
  • 初始化一個(gè)三維的numpy數(shù)組，用于存儲(chǔ)該分塊的所有波段和所有日期的數(shù)據(jù)。
  • 遍歷每個(gè)日期，使用readTif函數(shù)讀取該分塊的每個(gè)日期的每個(gè)波段的數(shù)據(jù)，并將其存儲(chǔ)到numpy數(shù)組中。
  • 打印寫入中的提示。
  • 遍歷每個(gè)像素，將其對(duì)應(yīng)的每個(gè)波段的每個(gè)日期的數(shù)據(jù)組合成一個(gè)字典，作為HBase表中的列值。使用分塊編號(hào)、行號(hào)和列號(hào)拼接成一個(gè)字符串，作為HBase表中的行鍵。使用put方法將行鍵和列值寫入HBase表中。
  • 關(guān)閉happybase連接對(duì)象。

def load(kuai,fenkuailist,datelist):
        connection = happybase.Connection('192.168.1.100')
        # # before first use:
        connection.open()
        table = connection.table('rawdata')
        
        print("(%d/%d)塊編號(hào)："%(kuai+1,len(fenkuailist)),fenkuailist[kuai])
        # 初始化立方體
        img_file = tifDir + "\\" + datelist[0] + "-" + fenkuailist[kuai] + ".tif"
        im_width, im_height, im_bands, im_data, kuai_im_geotrans, kuai_im_proj = readTif(img_file)

        tmpttt = np.empty((im_bands, im_width * im_height, len(datelist)))
    # print("波段 %s 個(gè)" % im_bands)
    # print("行列數(shù)", im_width, im_height)
        for shijian in range(len(datelist)):
        #  圖像
            img_file = tifDir + "\\" + datelist[shijian] + "-" + fenkuailist[kuai] + ".tif"
            # print(img_file)
            im_width, im_height, im_bands, im_data, im_geotrans, im_proj = readTif(img_file)
            kuai_im_geotrans = im_geotrans
            kuai_im_proj=im_proj

            for j in range(im_bands):
                tmpttt[j, :, shijian] = im_data[j].flatten(order='C')
        print("寫入中...")
        for index in tqdm(range(im_width * im_height)):
            dt={}
            for ban in range(im_bands):
                d1=zip(map(lambda  x:"f1:"+x+bandlist[ban],datelist),tmpttt[ban, index, :].astype(str))
            # Converting zip object to dict using dict() contructor.
                dt.update(d1)
            key=str(kuai%3)+fenkuailist[kuai][6:10]+fenkuailist[kuai][-4:]+str(index)
            table.put(key, dt)  # 提交數(shù)據(jù)，0001代表行鍵，寫入的數(shù)據(jù)要使用字典形式表示
        connection.close()  # 關(guān)閉傳輸

五、使用多線程技術(shù)

• 導(dǎo)入threading模塊，該模塊提供了多線程的支持，可以創(chuàng)建和管理線程，以及實(shí)現(xiàn)線程間的同步和通信。
• 使用time模塊的perf_counter函數(shù)，記錄程序運(yùn)行的開(kāi)始時(shí)間。例如：

# #計(jì)時(shí)開(kāi)始
start = time.perf_counter()

• 使用threading模塊的Semaphore類，創(chuàng)建一個(gè)信號(hào)量對(duì)象，用于限制線程的最大數(shù)量。信號(hào)量對(duì)象有一個(gè)內(nèi)部計(jì)數(shù)器，每當(dāng)一個(gè)線程調(diào)用acquire方法時(shí)，計(jì)數(shù)器減一，每當(dāng)一個(gè)線程調(diào)用release方法時(shí)，計(jì)數(shù)器加一。如果計(jì)數(shù)器為零，那么acquire方法將阻塞，直到其他線程調(diào)用release方法。例如：

sem=threading.Semaphore(5) #限制線程的最大數(shù)量為5

• 遍歷每個(gè)分塊，使用sem對(duì)象的acquire方法獲取一個(gè)信號(hào)量，然后使用threading模塊的Thread類，創(chuàng)建一個(gè)新的線程對(duì)象，并指定目標(biāo)函數(shù)為load函數(shù)，以及傳遞分塊編號(hào)、分塊列表和日期列表作為參數(shù)。然后調(diào)用start方法啟動(dòng)該線程。例如：

for kuai in range(30,72):
    sem.acquire()
    threading.Thread(target = load, args = (kuai,fenkuailist,datelist)).start()

• 在load函數(shù)中，在完成數(shù)據(jù)的讀取和寫入后，使用sem對(duì)象的release方法釋放一個(gè)信號(hào)量，并關(guān)閉happybase連接對(duì)象。例如：

def load(kuai,fenkuailist,datelist):
        connection = happybase.Connection('192.168.1.100')
        # # before first use:
        connection.open()
        table = connection.table('rawdata')
        
        print("(%d/%d)塊編號(hào)："%(kuai+1,len(fenkuailist)),fenkuailist[kuai])
        # 初始化立方體
        img_file = tifDir + "\\" + datelist[0] + "-" + fenkuailist[kuai] + ".tif"
        im_width, im_height, im_bands, im_data, kuai_im_geotrans, kuai_im_proj = readTif(img_file)

        tmpttt = np.empty((im_bands, im_width * im_height, len(datelist)))
    # print("波段 %s 個(gè)" % im_bands)
    # print("行列數(shù)", im_width, im_height)
        for shijian in range(len(datelist)):
        #  圖像
            img_file = tifDir + "\\" + datelist[shijian] + "-" + fenkuailist[kuai] + ".tif"
            # print(img_file)
            im_width, im_height, im_bands, im_data, im_geotrans, im_proj = readTif(img_file)
            kuai_im_geotrans = im_geotrans
            kuai_im_proj=im_proj

            for j in range(im_bands):
                tmpttt[j, :, shijian] = im_data[j].flatten(order='C')
        print("寫入中...")
        for index in tqdm(range(im_width * im_height)):
            dt={}
            for ban in range(im_bands):
                d1=zip(map(lambda  x:"f1:"+x+bandlist[ban],datelist),tmpttt[ban, index, :].astype(str))
            # Converting zip object to dict using dict() contructor.
                dt.update(d1)
            key=str(kuai%3)+fenkuailist[kuai][6:10]+fenkuailist[kuai][-4:]+str(index)
            table.put(key, dt)  # 提交數(shù)據(jù)，0001代表行鍵，寫入的數(shù)據(jù)要使用字典形式表示
        connection.close()  # 關(guān)閉傳輸
        sem.release()

• 使用time模塊的perf_counter函數(shù)，記錄程序運(yùn)行的結(jié)束時(shí)間，并計(jì)算程序運(yùn)行的總時(shí)間。例如：

# #計(jì)時(shí)結(jié)束
delta = time.perf_counter()-start
print("程序運(yùn)行的時(shí)間是:{}秒".format(delta))

到了這里，關(guān)于Python使用多線程操作tif影像和HBase數(shù)據(jù)庫(kù)的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！