使用happybase扫描远程hbase表时,会发生“tsocket read 0 bytes error”

eqzww0vc  于 2021-06-09  发布在  Hbase
关注(0)|答案(2)|浏览(540)

我正在尝试扫描具有超过100000000行的远程hbase表。扫描后,使用扫描行,尝试在hdfs中使用csv文件。
我试了将近三个星期来解决它,但我做不到。
用这种方法我扫描数据并制作csv文件
错误消息
/host/anaconda3/lib/python3.6/site-packages/thriftpy/transport/socket.py的源代码
/host/anaconda3/lib/python3.6/site-packages/thriftpy/transport/socket.py的源代码
==>我尝试了compat协议,增加了网络tcp内存缓冲区,增加了超时配置,在扫描参数中设置了1到10000个批量大小等等。。
但它几乎在30分钟内运行良好,但突然出现了错误。差不多1/50的时间它完成得很好。(工作很好没有任何错误)请帮助我。我试图找出错误的原因。但我拿不到。
有人知道怎么解决吗?
这是我的密码

import sys
print ("--sys.version--")
print (sys.version)
from pyhive import hive
import csv
import os
import happybase
import time
import subprocess
import datetime
import chardet
import logging
logging.basicConfig(level=logging.DEBUG)

csv_list=[]

col=[]
def conn_base():
    print('conn_base starts')

    #SETTING CONNECTION AND CONFIGURATION
    conn=happybase.Connection('13.xxx.xxx.xxx',port=9090)
    table=conn.table(b'TEMP_TABLE')

    #ITERATE DATA AND MAKE CSV FILE PER 100,000 RECORD. AND TAKE A TIME TO SLEEP PER 500000
    tmp=[]
    print('LET\'S MAKE CSV FILE FROM HBASE')
    index=0
    st=0
    global csv_list
    for row_key, data in table.scan():
        try:
           if (st%1000000==0):
                time.sleep(30)
                print("COUNT: ",st)
            if (st%500000==0):

               print("CHANGE CSV _FILE")
                index+=1
                ta_na='TEMP_TABLE'+str(index)+'_version.csv'
                csv_list.append(ta_na)

            st+=1
            with open('/home/host01/csv_dir/TEMP_TABLE/'+csv_list[index-1] ,'a') as f:
                tmp=[]
                tmp.append(data[b'CF1:XXXXX'].decode())
                tmp.append(data[b'CF1:YYYYY'].decode())
                tmp.append(data[b'CF1:DDDDD'].decode())
                tmp.append(data[b'CF1:SSSSS'].decode())
                tmp.append(data[b'CF1:GGGGG'].decode())
                tmp.append(data[b'CF1:HHHHH'].decode())
                tmp.append(data[b'CF1:QQQQQ'].decode())
                tmp.append(data[b'CF1:WWWWWW'].decode())
                tmp.append(data[b'CF1:EEEEE'].decode())
                tmp.append(data[b'CF1:RRRRR'].decode())

                f.write(",".join(tmp)+'\n')
                tmp=[]

        except:
            pass

        #PUT CSV FILES TO HDFS.
        st=1
        for i in range(len(csv_list)):
            try:
                st+=1
                cmd="hdfs dfs -put /home/host01/csv_dir/TEMP_TABLE"+str(csv_list[i])+" /user/hive/warehouse/TEMP_TABLE/"
                subprocess.call(cmd,shell=True)
                if (st%50==0):
                    time.sleep(5)

            except:
                pass
        cmd="hdfs dfs -put /home/host01/csv_dir/TEMP_TABLE/*.csv  /user/hive/warehouse/TEMP_TABLE/"
        subprocess.call(cmd,shell=True)

        print("PUT ALL CSV FILES TO HDFS")
        conn.close()
dw1jzc5e

dw1jzc5e1#

首先确保hbase thrift服务器已启动并正在运行。您可以使用以下命令运行thrift server:

hbase-daemon.sh start thrift [ -p 9090 ]

如果要指定端口号,请使用-p。默认端口为 9090

qc6wkl3g

qc6wkl3g2#

看上面的代码会让事情变得更复杂,这只是几个简单的步骤
确保hbase thrift已启动并正在运行。(使用上面的命令)
在hdfs设置文件中启用get-webhdfs。
从hdfs包使用不安全的客户机类(如果不是kerberos认证的)直接将文件写入hdfs(非常简单)

相关问题