Boy】

Boy#xff0c;不接受反驳#xff0c;毕竟大…从Hive源码解读大数据开发为什么可以脱离SQL、Java、Scala

【本文适合有一定计算机基础/半年工作经验的读者食用。

立个Flg愿天下不再有肤浅的SQL

Boy】

Boy不接受反驳毕竟大数据主要就是为机器学习和统计报表服务的自然从Oracle数据库开发转过来并且还是只会写几句SQL的人不在少数个别会Python写个spark.sql(“一个sql字符串”)的已经是SQL

Boy中的人才。

这种只能处理结构化表的最基础的大数据开发人员就是我们常提到的梗肤浅的SQL

Boy。

。

。

对大数据完全不懂思想还停留在数据库时代大数据组件也都是拿来当RDBMS来用。

。

。

这种业务开发人员的技术水平其实不敢恭维。

还有从Java后端开发转过来的虽然不适应但还是可以一个Main方法流畅地操作Spark、Flink手写个JDBC做点简单的二开这种就是平台开发人员技术水平要更高一些。

Java写得好Scala其实上手也快。

但是。

。

。

这并不代表做大数据只能用SQL/Java/Scala。

。

。

这么局限的话也不比SQL

笔者最早还搞过嵌入式开发自然明白C/C#/C也可以搞大数据。

。

。

table为例抛砖引玉解读为神马大数据开发可以脱离SQL、Java、Scala。

数据不外乎结构化数据和非结构化数据SQL只能处理极其有限的结构化表【RDBMS、整齐的csv/tsv等】绝大多数的半结构化、非结构化数据SQL是无能为力的【log日志文件、音图等】。

古代的MapReduce本身就不可以用SQLSpark和Flink老版本都是基于API的没有SQL的年代大家也活得好好的。

大数据组件对SQL的支持日渐友好都是后来的事情主要是为了降低门槛让SQL

Boy们当然只知道

db_name.tb_name;正常情况这个Hive表就会被drop掉认知也就局限于Hive是个数据库。

API

https://svn.apache.org/repos/infra/websites/production/hive/content/javadocs/r3.1.3/api/index.html

知道还有这种方式

org.apache.hadoop.hive.conf.HiveConf;

import

org.apache.hadoop.hive.metastore.HiveMetaStoreClient;/***

program:

HiveMetaStoreClient(hiveConf);client.dropTable(db_name,tb_name);}

}通过调用API的方式同样可以drop掉表。

显然不一定要用DDL。

通过HiveMetaStoreClient的方式还可以create建表等操作。

懂大数据底层的平台开发当然还有更狠的方式直接连Hive存元数据的MySQL对元数据表的数据做精准crud。

。

。

对结构化表的ETL或者其它的运算处理完全可以用Spark的DataFrame、Flink的DataStream编程纯API方式实现SQL能实现的Java和Scala都能实现至于SQL实现不了的Java和Scala也能实现。

。

。

笔者实在是想不到除了RDBMS和各类包皮产品【在开源的Apache组件基础上做一些封装】还有哪些场景是只能用SQL的。

。

。

虽然Hive底层是Java写的但是这并不意味着只能用Java操作Hive。

认知这么肤浅的话也就活该一辈子调参调API了。

。

。

Overridepublic

TException,NoSuchObjectException,

{dropTable(getDefaultCatalog(conf),

dbname,

{dropTable(getDefaultCatalog(conf),

dbname,

{dropTable(getDefaultCatalog(conf),

dbname,

warehouseOptions;warehouseOptions

new

HashMap();warehouseOptions.put(ifPurge,

TRUE);envContext

EnvironmentContext(warehouseOptions);}dropTable(catName,

dbName,

envContext);}虽然有多个同名方法但是底层调用的还是同一个方法

/***

org.apache.hadoop.hive.metastore.api.ThriftHiveMetastore.Iface#drop_table(java.lang.String,*

java.lang.String,

TException,NoSuchObjectException,

{Table

{hook.preDropTable(tbl);}boolean

success

{drop_table_with_environment_context(catName,

dbname,

TRUE.equals(envContext.getProperties().get(ifPurge))));}successtrue;}

catch

{hook.rollbackDropTable(tbl);}}}主要就是获取了表对象然后做了preDropTable预提交和commitDropTable实际的提交。

这种2PC方式表面上还是很严谨。

。

。

package

org.apache.hadoop.hive.metastore;/***

HiveMetaHook

MetaStoreUtils#isExternalTable}

to*

org.apache.hadoop.hive.metastore;public

abstract

org.apache.hadoop.hive.ql.lockmgr.TestDbTxnManager2#testLockingOnInsertIntoNonNativeTables()}*/

public

org.apache.hadoop.hive.accumulo;/***

Create

HiveMetaHook,HiveStoragePredicateHandler

Overridepublic

connectionParams.getConnector().tableOperations();if

catch

MetaException(StringUtils.stringifyException(e));}

catch

MetaException(StringUtils.stringifyException(e));}

catch

MetaException(StringUtils.stringifyException(e));}}}按照最简单的内部表、需要删数据来看实际上调用的是这个delete方法。

而TableOperations又是个接口

package

org.apache.accumulo.core.client.admin;/***

Provides

org.apache.accumulo.core.client.impl;public

class

Arrays.asList(ByteBuffer.wrap(tableName.getBytes(UTF_8)));MapString,String

opts

{doTableFateOperation(tableName,

args,

}所以实际入口是这里的doTableFateOperation方法。

枚举体的FateOperation.TABLE_DELETE2。

跳转到

namespaceNotFoundExceptionClass,

FateOperation

AccumuloSecurityException,TableExistsException,

AccumuloException,

beginFateOperation();executeFateOperation(opid,

op,

waitForFateOperation(opid);return

ret;}

NamespaceNotFoundException(null,

Target

Tables.getPrintableTableInfoFromName(context.getInstance(),

new

AccumuloSecurityException(e.user,

e.code,

NamespaceExistsException(e);case

new

NamespaceNotFoundException(e);case

OFFLINE:throw

TableOfflineException(context.getInstance(),

Tables.getTableId(context.getInstance(),

tableOrNamespaceName));default:throw

new

AccumuloException(e.description,

e);}}

AccumuloException(e.getMessage(),

e);}

{Tables.clearCache(context.getInstance());//

always

e);}}}在这里可以发现一些奇怪的现象居然catch了好多Thrift相关的Exception。

继续跳转

This

MasterClient.getConnectionWithRetry(context);client.executeFateOperation(Tracer.traceInfo(),

context.rpcCreds(),

tte);UtilWaitThread.sleep(100);}

finally

{MasterClient.close(client);}}}这个死循环里获取了Client对象。

但是这个Client一看就没那么简单。

。

。

调用的executeFateOperation方法还不能直接Idea点开需要手动定位。

分析client对象

org.apache.accumulo.core.client.impl;import

com.google.common.net.HostAndPort;public

class

LoggerFactory.getLogger(MasterClient.class);public

static

getConnectionWithRetry(ClientContext

context)

result;UtilWaitThread.sleep(250);}}

public

org.apache.accumulo.core.master.thrift;SuppressWarnings({unchecked,

serial,

executeFateOperation(org.apache.accumulo.core.trace.thrift.TInfo

tinfo,

org.apache.accumulo.core.security.thrift.TCredentials

credentials,

org.apache.accumulo.core.client.impl.thrift.ThriftSecurityException,

org.apache.accumulo.core.client.impl.thrift.ThriftTableOperationException,

org.apache.thrift.TException;}public

void

executeFateOperation(org.apache.accumulo.core.trace.thrift.TInfo

tinfo,

org.apache.accumulo.core.security.thrift.TCredentials

credentials,

org.apache.accumulo.core.client.impl.thrift.ThriftSecurityException,

org.apache.accumulo.core.client.impl.thrift.ThriftTableOperationException,

{send_executeFateOperation(tinfo,

credentials,

autoClean);recv_executeFateOperation();

}public

org.apache.thrift.TServiceClient

implements

}所以这种client对象才可以执行executeFateOperation方法。

分为2步字面意思send_executeFateOperation方法发送了啥recv_executeFateOperation方法又接收了啥。

显然发送消息是需要重点关心的

public

send_executeFateOperation(org.apache.accumulo.core.trace.thrift.TInfo

tinfo,

org.apache.accumulo.core.security.thrift.TCredentials

credentials,

executeFateOperation_args();args.setTinfo(tinfo);args.setCredentials(credentials);args.setOpid(opid);args.setOp(op);args.setArguments(arguments);args.setOptions(options);args.setAutoClean(autoClean);sendBase(executeFateOperation,

args);

}这个发送的方法把入参的表名、操作类型【Drop表】设置为sendBase方法的入参。

package

seqid_));args.write(oprot_);oprot_.writeMessageEnd();oprot_.getTransport().flush();}}其中

package

org.apache.thrift.protocol;/***

Message

org.apache.thrift.protocol;/***

Helper

org.apache.thrift.protocol.TProtocol;/***

Generic

}按照注释可以知道write方法是把Java的对象输出给协议。

public

org.apache.thrift.TBaseexecuteFateOperation_args,

executeFateOperation_args._Fields,

Cloneable,

ComparableexecuteFateOperation_args

public

write(org.apache.thrift.protocol.TProtocol

oprot)

{schemes.get(oprot.getScheme()).getScheme().write(oprot,

this);}

org.apache.thrift.scheme;import

interface

read(org.apache.thrift.protocol.TProtocol

iproto,

org.apache.thrift.TException;public

void

write(org.apache.thrift.protocol.TProtocol

oproto,

org.apache.thrift.TException;}又是跳转到接口。

。

。

package

org.apache.thrift.protocol;import

import

org.apache.thrift.scheme.IScheme;

import

org.apache.thrift.scheme.StandardScheme;

import

org.apache.thrift.transport.TTransport;/***

Protocol

}显然get到的是StandardScheme类。

而writeMessageBegin又是这个抽象类的抽象方法。

至此可以知道原生支持的协议有这些。

最常用的当然就是二进制协议TBinaryProtocol。

package

org.apache.thrift.protocol;import

java.io.UnsupportedEncodingException;

import

org.apache.thrift.transport.TTransport;/***

Binary

message.type;writeI32(version);writeString(message.name);writeI32(message.seqid);}

else

{writeString(message.name);writeByte(message.type);writeI32(message.seqid);}}

}可以看出writeMessageBegin方法就是实际的写数据操作把消息拆分后写出。

public

str.getBytes(UTF-8);writeI32(dat.length);trans_.write(dat,

dat.length);}

org.apache.thrift.transport;import

Generic

搞过嵌入式开发的一定很熟悉这个Socket就是IPport的那个Socket。

应用层与TCP/IP传输层间的抽象层。

。

。

package

org.apache.thrift.transport;/***

This

TTransportException(TTransportException.NOT_OPEN,

Cannot

TTransportException(TTransportException.UNKNOWN,

iox);}if

TTransportException(TTransportException.END_OF_FILE);}return

bytesRead;}/***

TTransportException(TTransportException.NOT_OPEN,

Cannot

TTransportException(TTransportException.UNKNOWN,

iox);}}/***

TTransportException(TTransportException.NOT_OPEN,

Cannot

TTransportException(TTransportException.UNKNOWN,

iox);}}

顶层API【dropTable】→表操作实现类【TableOperationsImpl】的删表方法【doTableFateOperation】

→executeFateOperation方法→Client类的实例对象的executeFateOperation方法

→sendBase方法→executeFateOperation_args静态类的实例对象的write方法输出数据给传输协议TProtocol

→传输协议类的write方法具体把数据写出给Thrift的Server

→Thrift的Server接收到消息后执行对应的操作最出名的Thrift当然是Hive自己的Hive

Server借助它们可以用JDBC或者Cli的方式去操作Hive。

但是Thrift的初衷就是实现语言无关毕竟底层只需要能把数据传输到位即可数据传输并不是Java的特权。

在service-rpc这个路径下可以发现有cpp、Java、php、pyrb的包

https://cwiki.apache.org/confluence/display/Hive/HiveClient#HiveClient-ThriftJavaClient

The

命令行模式目前只能用于嵌入式服务JDBC和Thrift-Java的Client可以支持嵌入式和独立部署的服务。

别的语言的Client只支持在独立部署的服务使用。

Connection

DriverManager.getConnection(jdbc:hive://localhost:10000/default,

Statement

con.createStatement();这种古代的Hive

Connection

DriverManager.getConnection(jdbc:hive2://localhost:10000/default,

);这种Hive

TTransport.TBufferedTransport(transport)protocol

TBinaryProtocol.TBinaryProtocol(transport)client

ThriftHive.Client(protocol)transport.open()client.execute(CREATE

TABLE

client.fetchAll()transport.close()except

Thrift.TException,

packages/hive_service/ThriftHive.php;

require_once

$transport-close();Ruby好歹也给了个参考

https://github.com/forward3d/rbhive

至于Java、C就不给Client的案例了。

。

。

也是很容易理解。

。

。

毕竟Java有JDBC和高层API一般不会有人去用底层API了。

如果是做平台开发或者组件开发这种真正用得上底层API的情况地方支援中央发型的老Java程序猿查API填参数让程序跑起来这点工程能力还是有的。

既然可以通过Thrift实现语言无关那么调用组件就不必局限于Java或者Scala。

而造轮子从来也不是Java和Scala的专利。

大数据并不是趋向SQL化只是为了扩大受众群体让广大技术水平不高的业务开发人员也能吃上大数据技术的红利。

且SQL在处理结构化表的特定场景下开发效率更高。

但是。

。

。

哪怕是这种极度细分的场景SQL还是有很多缺陷虽然API的方式也没有好到哪里去。

造轮子和组件调用就更是语言无关的事情了。

。

。

编程语言往往只是个表达思想的载体技术栈足够全面才有做选择的权力。

转载请注明出处https://lizhiyong.blog.csdn.net/article/details/129742904