Hive——UDF函数：高德地图API逆地理编码，实现离线解析经纬度转换省市区（离线地址库，非调用高德API）

1. 需求背景

数据现状

目前业务系统某数据库表中记录了约3亿条用户行为数据，其中两列记录了用户触发某个业务动作时所在的经度和纬度数值，但是没有记录经纬度对应的省市区编码和名称信息。

业务需求

现在业务方提出一个数据需求，想要统计省市区对应的用户数等相关指标。

面临技术问题

因为历史数据量较大，如果通过调用高德API把所有历史数据中的经纬度对应的省市区请求回来，会面临一个个问题：在查看公司的高德API账户后，发现每天提供的最大调用量是300W次，那么要把历史3亿数据初始化调用完，需要30000W/300w=100天，要3个多月，这完全是不可接受的。

寻求其他方案

既然不能通过调用高德API的方式获取省市区，那有没有一个离线的地址库，然后从这个地址库获取历史数据的省市区呢。然后通过搜索引擎还真的找到了某个大神写的一个第三方库，可以从一个地址库文件来获取经纬度对应的省市区。
这个第三方库的Github地址：https://github.com/hsp8712/addrparser

2. 运行环境

软件版本

• Java 1.8
• Hive 3.1.0
• Hadoop 3.1.1

Maven依赖

        <dependency><groupId>tech.spiro</groupId><artifactId>addrparser</artifactId><version>1.1</version></dependency><dependency><groupId>org.apache.hive</groupId><artifactId>hive-exec</artifactId><version>3.1.0</version><scope>provided</scope></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-common</artifactId><version>3.1.1</version><scope>provided</scope></dependency>

3. 获取离线地址库

这个第三方库的作者提供了一个2019年9月份的离线地址库文件，考虑到这个文件的数据已经比较旧了，然后去翻看作者的源码文件，发现提供了爬取地址库的源码，直接拿来改改就可以用了（但前提是你要有请求高德API的API Key）。

下面是本人调整后并经过测试后可以正常请求地址库的代码（这个类中还引用了作者编写的其他类/接口，请参考作业的Github）：

import org.apache.commons.cli.*;
import tech.spiro.addrparser.crawler.GetRegionException;
import tech.spiro.addrparser.crawler.RegionDataCrawler;
import tech.spiro.addrparser.io.RegionDataOutput;
import tech.spiro.addrparser.io.file.JSONFileRegionDataOutput;import java.io.IOException;
import java.util.Arrays;/*
*
* A command-line tool to crawl region data.
* */
public class CrawlerServer {private static Options options = new Options();static {options.addOption("k", "key", true, "Amap enterprise dev key");options.addOption("l", "level", true, "Root region level: 0-country, 1-province, 2-city");options.addOption("c", "code", true, "Root region code");options.addOption("o", "out", true, "Output file.");}private static void printHelp() {HelpFormatter formatter = new HelpFormatter();formatter.printHelp("CrawlerServer", options );}public static void main(String[] args) throws IOException, GetRegionException {CommandLineParser parser = new BasicParser();try {CommandLine cmd = parser.parse(options, args);
//            String key  = cmd.getOptionValue("k");
//            String level = cmd.getOptionValue('l');
//            String code = cmd.getOptionValue('c');
//            String outputFile = cmd.getOptionValue('o');String key  = "xxxxxxxxxxxxxxxxxxx";String level = "0";String code = "100000";String outputFile = "/Users/name/Desktop/china-region.json";if (!Arrays.asList("0", "1", "2").contains(level)) {throw new ParseException("option:level invalid.");}int _code = 0;try {_code = Integer.parseInt(code);} catch (NumberFormatException e) {throw new ParseException("code must be numeric.");}execute(key, level, _code, outputFile);} catch (ParseException e) {System.out.println(e.getMessage());printHelp();System.exit(-1);}}private static void execute(String amapKey, String level, int code, String out) throws IOException, GetRegionException {try (RegionDataOutput regionOutput = new JSONFileRegionDataOutput(out)) {RegionDataCrawler infoLoader = new RegionDataCrawler(regionOutput, amapKey);if ("0".equals(level)) {infoLoader.loadCountry();} else if ("1".equals(level)) {infoLoader.loadProv(code);} else if ("2".equals(level)) {infoLoader.loadCity(code);}}}
}

运行上面这段代码获取全国省市区地址库实践会比较久，大概要30分钟左右，生产的json文件china-region.json大小约160M。

4. Hive UDF函数实现

在获取到地址库数据之后，为了实现输入经度、维度输出省市区编码和名称的UDF函数，我们需要先把这个地址库文件china-region.json上传到一个指定的HDFS目录下面，这样在Hive中使用UDF函数的时候可以从HDFS目录下直接查询这个文件。

下面代码就是UDF函数的实现逻辑：输入经度、维度数值，然后查询离线地址库文件china-region.json，最终输出对应的省市区信息的json字符串。

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.ql.exec.Description;
import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDF;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;
import org.apache.hadoop.io.Text;
import tech.spiro.addrparser.common.RegionInfo;
import tech.spiro.addrparser.io.RegionDataInput;
import tech.spiro.addrparser.io.file.JSONFileRegionDataInput;
import tech.spiro.addrparser.parser.Location;
import tech.spiro.addrparser.parser.LocationParserEngine;
import tech.spiro.addrparser.parser.ParserEngineException;import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;@Description(name = "GetRegionInfo",value = "_FUNC_(latitude, longitude) - Returns the province, city, and district names and codes based on latitude and longitude"
)
public class LgtLttUDF extends GenericUDF {// 经纬度-省市区基础库文件private static final String RESOURCE_FILE = "hdfs://nameservice/user/username/udf/china-region.json";// 位置解析引擎private static volatile LocationParserEngine sharedEngine;private static final Object lock = new Object();/*** 1. UDF函数入参校验* 2. 创建并初始化位置解析引擎* 3. 设置UDF函数返回值数据类型* @param arguments* @return ObjectInspector* @throws UDFArgumentException*/@Overridepublic ObjectInspector initialize(ObjectInspector[] arguments) throws UDFArgumentException {// 参数数量校验if (arguments.length != 2) {throw new UDFArgumentException("The function requires two arguments.");}// 参数类型校验if (!arguments[0].getCategory().equals(ObjectInspector.Category.PRIMITIVE) ||!arguments[1].getCategory().equals(ObjectInspector.Category.PRIMITIVE)) {throw new UDFArgumentException("GetRegionInfoUDF only accepts primitive types as arguments.");}// 创建并初始化位置解析引擎initializeSharedEngine();// 返回值return ObjectInspectorFactory.getStandardStructObjectInspector(Arrays.asList("province_name", "province_code", "city_name", "city_code", "district_name", "district_code"),Arrays.asList(PrimitiveObjectInspectorFactory.javaStringObjectInspector,PrimitiveObjectInspectorFactory.javaStringObjectInspector,PrimitiveObjectInspectorFactory.javaStringObjectInspector,PrimitiveObjectInspectorFactory.javaStringObjectInspector,PrimitiveObjectInspectorFactory.javaStringObjectInspector,PrimitiveObjectInspectorFactory.javaStringObjectInspector));}/*** 初始化位置解析引擎* @throws UDFArgumentException*/private void initializeSharedEngine() throws UDFArgumentException {if (sharedEngine == null) {synchronized (lock) {if (sharedEngine == null) {try {// china-region.json文件作为基础数据InputStreamReader reader = getJsonFileInputStreamFromHDFS(RESOURCE_FILE);RegionDataInput regionDataInput = new JSONFileRegionDataInput(reader);// 创建位置解析引擎sharedEngine = new LocationParserEngine(regionDataInput);// 初始化，加载数据，比较耗时sharedEngine.init();} catch (ParserEngineException | IOException e) {throw new UDFArgumentException("Failed to initialize LocationParserEngine: " + e.getMessage());}}}}}/*** 从HDFS路径读取JSON文件并返回InputStreamReader。** @param hdfsPath HDFS上的JSON文件路径* @return 文件的InputStreamReader对象，用于进一步读取内容* @throws IOException 如果发生I/O错误*/public static InputStreamReader getJsonFileInputStreamFromHDFS(String hdfsPath) throws IOException {// 创建Hadoop配置对象Configuration conf = new Configuration();// 根据配置获取文件系统实例FileSystem fs = FileSystem.get(conf);// 构建HDFS路径对象Path path = new Path(hdfsPath);// 检查文件是否存在if (!fs.exists(path)) {throw new IOException("File " + hdfsPath + " does not exist on HDFS.");}// 打开文件并获取输入流return new InputStreamReader(fs.open(path), "UTF-8");}/*** 通过经度、维度获取对应省市区的编码和名称* @param args* @return* @throws HiveException*/@Overridepublic Object evaluate(DeferredObject[] args) throws HiveException {if (args == null || args.length != 2) {return null;}try {// 经度double longitude = Double.parseDouble(args[1].get().toString());// 纬度double latitude = Double.parseDouble(args[0].get().toString());// 位置信息Location location = sharedEngine.parse(latitude, longitude);// 省市区信息RegionInfo provInfo = location.getProv();RegionInfo cityInfo = location.getCity();RegionInfo districtInfo = location.getDistrict();// 返回省市区编码、名称return new Object[]{new Text(provInfo.getName()),new Text(String.valueOf(provInfo.getCode())),new Text(cityInfo.getName()),new Text(String.valueOf(cityInfo.getCode())),new Text(districtInfo.getName()),new Text(String.valueOf(districtInfo.getCode()))};} catch (Exception e) {throw new HiveException("Error processing coordinates", e);}}@Overridepublic String getDisplayString(String[] children) {return "ltt_lgt_region(" + children[0] + ", " + children[1] + ")";}
}

5. 创建Hive UDF函数

将第四步的UDF实现代码打jar包：china_region.jar，Hive和Hadoop依赖不需要打进去，因为集群上都是有的，只需要把这个第三方的addrparser打进去就可以了。

1. 将jar包china_region.jar上传到HDFS指定目录下

2. 将jar包添加到hive的classpath。在Hive的cli中执行如下命令

   hive> add jar china_region.jar
   

3. 创建UDF函数（永久性UDF函数）

   hive> create function ltt_lgt as 'com.hive.udf.LgtLttUDF ' using jar 'hdfs://nameservice/user/username/udf/china_region.jar';
   

4. 测试UDF函数

   hive> select ltt_lgt(100.750934, 26.038634)
   

6. 参考

1. https://github.com/hsp8712/addrparser