I have HDFS running and am trying to join a table from there to a feature class in ArcGIS Desktop. I can do any query I want inside the Cloudera Hue browser with the table, but I can not get it to join to my feature class in ArcGIS. I keep receiving an error that the table does not exist/is not supported. I am trying to join a ziptrips table (that has zip codes with population) to a feature class called nyc_zip. My apologies for any messy code, I have butchered this thing so far.
import arcpy
import os
import re
import requests
from pyhive import hive
import contextlib
class WebHDFSRes(object):
def __init__(self, res):
self.res = res
def __enter__(self):
return self.res
def __exit__(self, exception_type, exception_value, traceback):
self.res.close()
class WebHDFS(object):
def __init__(self, host, user, port=50070):
self.host = host
self.port = port
self.user = user
def open(self, hdfs_path, offset=-1, length=-1, buffer_size=-1):
params = {'op': 'OPEN', 'user.name': self.user}
if offset > 0:
params['offset'] = offset
if length > 0:
params['length'] = length
if buffer_size > 0:
params['buffersize'] = buffer_size
url = 'http://{}:{}/webhdfs/v1{}'.format(self.host, self.port, hdfs_path)
return WebHDFSRes(requests.get(url, params=params, stream=True))
def list_status(self, hdfs_path, suffix_re='*'):
files = []
prog = re.compile(suffix_re)
params = {'op': 'LISTSTATUS', 'user.name': self.user}
url = 'http://{}:{}/webhdfs/v1{}'.format(self.host, self.port, hdfs_path)
with WebHDFSRes(requests.get(url, params=params)) as res:
doc = res.json()
for i in doc['FileStatuses']['FileStatus']:
path_suffix = i['pathSuffix']
if prog.match(path_suffix):
files.append('{}/{}'.format(hdfs_path, path_suffix))
return files
class Toolbox(object):
def __init__(self):
self.label = 'ZipTool'
self.alias = 'ZipTool'
self.tools = [Tool]
class Tool(object):
def __init__(self):
self.label = 'TripsTool'
self.description = 'TripsTool'
self.canRunInBackground = True
def getParameterInfo(self):
paramFC = arcpy.Parameter(
name='out_fc',
displayName='out_fc',
direction='Output',
datatype='Feature Layer',
parameterType='Derived')
# paramFC.symbology = os.path.join(os.path.dirname(__file__), 'GPSTool.lyr')
paramName = arcpy.Parameter(
name='in_name',
displayName='Name',
direction='Input',
datatype='GPString',
parameterType='Required')
paramName.value = 'TripsPoints'
paramHost = arcpy.Parameter(
name='in_host',
displayName='HDFS Host',
direction='Input',
datatype='GPString',
parameterType='Required')
paramHost.value = 'quickstart'
paramUser = arcpy.Parameter(
name='in_user',
displayName='User name',
direction='Input',
datatype='GPString',
parameterType='Required')
paramUser.value = 'root'
paramPath = arcpy.Parameter(
name='in_path',
displayName='HDFS Path',
direction='Input',
datatype='GPString',
parameterType='Required')
paramPath.value = '/user/root/'
paramFile = arcpy.Parameter(
name='in_file',
displayName='HDFS File(s)',
direction='Input',
datatype='GPString',
parameterType='Required')
paramFile.value = 'ziptrips'
return [paramFC, paramName, paramHost, paramUser, paramPath, paramFile]
def isLicensed(self):
return True
def updateParameters(self, parameters):
return
def updateMessages(self, parameters):
return
def insert_row(self, cursor, line):
try:
zipcode = [0]
pop = [1]
cursor.insertRow(zipcode, pop)
except Exception as e:
arcpy.AddWarning(str(e))
def execute(self, parameters, messages):
name = parameters[1].value
host = parameters[2].value
user = parameters[3].value
path = parameters[4].value
fext = parameters[5].value
in_memory = True
if in_memory:
ws = 'in_memory'
fc = ws + '/' + name
else:
fc = os.path.join(arcpy.env.scratchGDB, name)
ws = os.path.dirname(fc)
if arcpy.Exists(fc):
arcpy.management.Delete(fc)
sp_ref = arcpy.SpatialReference(4326)
arcpy.env.workspace = "C:/Data/BigData/HV_Spatial"
inFeature = "hivespatial.gdb/NYC_zips"
inLayer = "nyc_lyr"
inField = "ZCTA5CE10"
joinFeature = parameters[5].value
joinField = joinFeature[0]
outFC = "join"
arcpy.MakeFeatureLayer_management(inFeature, inLayer)
arcpy.AddJoin_management(inLayer, inField, joinFeature, joinField)
arcpy.CopyFeatures_management(inLayer, outFC)
arcpy.management.CreateFeatureclass(ws, name, 'POLYGON', spatial_reference=sp_ref)
arcpy.management.AddField(fc, 'Zip_Code', 'String', field_alias='Zip')
arcpy.management.AddField(fc, 'POP', 'String', field_alias='POP')
if in_memory:
arcpy.AddMessage("Connecting...")
with contextlib.closing(hive.connect(host, username=user, auth="NOSASL")) as conn:
cursor = conn.cursor()
if in_memory:
arcpy.AddMessage("Executing...")
cursor.execute(hql)
with arcpy.da.InsertCursor(fc, ["Zip" "POP"]) as insert_cursor:
if in_memory:
arcpy.AddMessage("Fetching...")
for row in cursor.fetchall():
insert_cursor.insertRow(row)
parameters[0].value = fc
