pyspark访问hive数据实战-低调大师

pyspark访问hive数据实战

2017-07-31 596

数据分析都是直接使用hive脚本进行调用，随着APP用户行为和日志数据量的逐渐累积，跑每天的脚本运行需要花的时间越来越长，虽然进行了sql优化，但是上spark已经提上日程。

直接进行spark开发需要去学习scala，为了降低数据分析师的学习成本，决定前期先试用sparkSQL，能够让计算引擎无缝从MR切换到spark，现在主要使用pyspark访问hive数据。

以下是安装配置过程中的详细步骤：

1.安装spark

需要先安装JDK和scala，这不必多说，由于现有hadoop集群版本是采用的2.6.3，所以spark版本是下载的稳定版本spark-1.4.0-bin-hadoop2.6.tgz

我是先在一台机器上完成了Spark的部署，Master和Slave都在一台机器上。注意要配置免秘钥ssh登陆。

1.1 环境变量配置

export JAVA_HOME=/usr/jdk1.8.0_73
export HADOOP_HOME=/usr/hadoop
export HADOOP_CONF_DIR=/usr/hadoop/etc/hadoop
export SCALA_HOME=/usr/local/scala-2.11.7
export SPARK_HOME=/home/hadoop/spark_folder/spark-1.4.0-bin-hadoop2.6
export SPARK_MASTER_IP=127.0.0.1
export SPARK_MASTER_PORT=7077
export SPARK_MASTER_WEBUI_PORT=8099
export SPARK_WORKER_CORES=3 //每个Worker使用的CPU核数
export SPARK_WORKER_INSTANCES=1 //每个Slave中启动几个Worker实例
export SPARK_WORKER_MEMORY=10G //每个Worker使用多大的内存
export SPARK_WORKER_WEBUI_PORT=8081 //Worker的WebUI端口号
export SPARK_EXECUTOR_CORES=1 //每个Executor使用使用的核数
export SPARK_EXECUTOR_MEMORY=1G //每个Executor使用的内存
export HIVE_HOME=/home/hadoop/hive
export SPARK_CLASSPATH=$HIVE_HOME/lib/mysql-connector-java-5.1.31-bin.jar:$SPARK_CLASSPATH
export LD_LIBRARY_PATH=${LD_LIBRARY_PATH}:$HADOOP_HOME/lib/native

1.2 配置slaves

cp slaves.template slaves
vi slaves 添加以下内容：localhost

1.3 启动master和slave

cd $SPARK_HOME/sbin/
./start-master.sh
启动日志位于 $SPARK_HOME/logs/目录，访问 http://localhost:8099，即可看到Spark的WebUI界面
执行 ./bin/spark-shell,打开Scala到Spark的连接窗口

2.SparkSQL与Hive的整合

拷贝$HIVE_HOME/conf/hive-site.xml和hive-log4j.properties到 $SPARK_HOME/conf/
在$SPARK_HOME/conf/目录中，修改spark-env.sh，添加
export HIVE_HOME=/home/hadoop/hive
export SPARK_CLASSPATH=$HIVE_HOME/lib/mysql-connector-java-5.1.31-bin.jar:$SPARK_CLASSPATH
另外也可以设置一下Spark的log4j配置文件，使得屏幕中不打印额外的INFO信息(如果不想受干扰可设置为更高):
log4j.rootCategory=WARN, console
进入$SPARK_HOME/bin，执行 ./spark-sql –master spark://127.0.0.1:7077 进入spark-sql CLI:
[hadoop@hadoop spark]$ bin/spark-sql --help
Usage: ./bin/spark-sql [options] [cli option]
CLI options:
-d,--define <keykey=value> Variable subsitution to apply to hive
commands. e.g. -d A=B or --define A=B
--database <databasename> Specify the database to use
-e <quoted-query-string> SQL from command line
-f <filename> SQL from files
-h <hostname> connecting to Hive Server on remote host
--hiveconf <propertyproperty=value> Use value for given property
--hivevar <keykey=value> Variable subsitution to apply to hive
commands. e.g. --hivevar A=B
-i <filename> Initialization SQL file
-p <port> connecting to Hive Server on port number
-S,--silent Silent mode in interactive shell
-v,--verbose Verbose mode (echo executed SQL to the
console)

需要注意的是CLI不是使用JDBC连接，所以不能连接到ThriftServer;但可以配置conf/hive-site.xml连接到hive的metastore，然后对hive数据进行查询。下面我们接着说如何在python中连接hive数据表查询。

3.配置pyspark和示例代码

3.1 配置pyspark

打开/etc/profile:
#PythonPath 将Spark中的pySpark模块增加的Python环境中
export PYTHONPATH=/opt/spark-hadoop/python
source /etc/profile

执行./bin/pyspark ,打开Python到Spark的连接窗口，确认没有报错。

打开命令行窗口，输入python,Python版本为2.7.6，如图所示，注意Spark暂时不支持Python3。输入import pyspark不报错，证明开发前工作已经完成。

3.2 启动ThriftServer

启动ThriftServer，使之运行在spark集群中：

sbin/start-thriftserver.sh --master spark://localhost:7077 --executor-memory 5g

ThriftServer可以连接多个JDBC/ODBC客户端，并相互之间可以共享数据。

3.3 请求示例

查看spark官方文档说明，spark1.4和2.0对于sparksql调用hive数据的API变化并不大。都是用sparkContext 。

from pyspark import SparkConf, SparkContext
from pyspark.sql import HiveContext
conf = (SparkConf()
.setMaster("spark://127.0.0.1:7077")
.setAppName("My app")
.set("spark.executor.memory", "1g"))
sc = SparkContext(conf = conf)
sqlContext = HiveContext(sc)
my_dataframe = sqlContext.sql("Select count(1) from logs.fmnews_dim_where")
my_dataframe.show()

返回结果：

运行以后在webUI界面看到job运行详情。

4.性能比较

截取了接近一个月的用户行为数据，数据大小为2G，总共接近1600w条记录。

为了测试不同sql需求情况下的结果，我们选取了日常运行的2类sql：

1.统计数据条数：

select count(1) from fmnews_user_log2;

2.统计用户行为：

SELECT device_id, min_time FROM
(SELECT device_id,min(import_time) min_time FROM fmnews_user_log2
GROUP BY device_id)a
WHERE from_unixtime(int(substr(min_time,0,10)),'yyyy-MM-dd') = '2017-03-02';

3. 用户行为分析：

select case when from_unixtime(int(substr(fmnews_time,0,10)),'HH:mm') between '06:00' and '07:59' then 1
when from_unixtime(int(substr(fmnews_time,0,10)),'HH:mm') between '08:00' and '09:59' then 2
when from_unixtime(int(substr(fmnews_time,0,10)),'HH:mm') between '10:00' and '11:59' then 3
when from_unixtime(int(substr(fmnews_time,0,10)),'HH:mm') between '12:00' and '13:59' then 4
when from_unixtime(int(substr(fmnews_time,0,10)),'HH:mm') between '14:00' and '15:59' then 5
when from_unixtime(int(substr(fmnews_time,0,10)),'HH:mm') between '16:00' and '17:59' then 6
when from_unixtime(int(substr(fmnews_time,0,10)),'HH:mm') between '18:00' and '19:59' then 7
when from_unixtime(int(substr(fmnews_time,0,10)),'HH:mm') between '20:00' and '21:59' then 8
when from_unixtime(int(substr(fmnews_time,0,10)),'HH:mm') between '22:00' and '23:59' then 9
else 0 end fmnews_time_type, count(distinct device_id) device_count,count(1) click_count
from fmcm.fmnews_user_log2
where from_unixtime(int(substr(import_time,0,10)),'yyyy-MM-dd') = '2017-03-02'
group by case when from_unixtime(int(substr(fmnews_time,0,10)),'HH:mm') between '06:00' and '07:59' then 1
when from_unixtime(int(substr(fmnews_time,0,10)),'HH:mm') between '08:00' and '09:59' then 2
when from_unixtime(int(substr(fmnews_time,0,10)),'HH:mm') between '10:00' and '11:59' then 3
when from_unixtime(int(substr(fmnews_time,0,10)),'HH:mm') between '12:00' and '13:59' then 4
when from_unixtime(int(substr(fmnews_time,0,10)),'HH:mm') between '14:00' and '15:59' then 5
when from_unixtime(int(substr(fmnews_time,0,10)),'HH:mm') between '16:00' and '17:59' then 6
when from_unixtime(int(substr(fmnews_time,0,10)),'HH:mm') between '18:00' and '19:59' then 7
when from_unixtime(int(substr(fmnews_time,0,10)),'HH:mm') between '20:00' and '21:59' then 8
when from_unixtime(int(substr(fmnews_time,0,10)),'HH:mm') between '22:00' and '23:59' then 9
else 0 end;

第一条sql的执行结果对比：hive 35.013 seconds

第一条sql的执行结果对比：sparksql 1.218 seconds

第二条sql的执行结果对比：hive 78.101 seconds

第二条sql的执行结果对比：sparksql 8.669 seconds

第三条sql的执行结果对比：hive 101.228 seconds

第三条sql的执行结果对比：sparksql 14.221 seconds

可以看到，虽然没有官网吹破天的100倍性能提升，但是根据sql的复杂度来看10~30倍的效率还是可以达到的。

不过这里要注意到2个影响因子：

1. 我们数据集并没有采取全量，在数据量达到TB级别两者的差距应该会有所减小。同时sql也没有针对hive做优化。

2. spark暂时是单机(内存足够)并没有搭建集群，hive使用的hadoop集群有4台datanode。

本文作者：aibati2008

来源：51CTO

微信关注我们

原文链接：https://yq.aliyun.com/articles/181222

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

给Java开发者的10个大数据工具和框架

当今IT开发人员面对的最大挑战就是复杂性，硬件越来越复杂，OS越来越复杂，编程语言和API越来越复杂，我们构建的应用也越来越复杂。根据外媒的一项调查报告，中软卓越专家列出了Java程序员在过去12个月内一直使用的一些工具或框架，或许会对你有意义。先来看看大数据的概念。根据维基百科，大数据是庞大或复杂的数据集的广义术语，因此传统的数据处理程序不足以支持如此庞大的体量。在许多情况下，使用SQL数据库存储/检索数据都是很好的选择。而现如今的很多情况下，它都不再能满足我们的目的，这一切都取决于用例的变化。现在来讨论一些不同的非SQL存储/处理数据工具，例如，NoSQL数据库，全文搜索引擎，实时流式处理，图形数据库等。 1、MongoDB——最受欢迎的，跨平台的，面向文档的数据库。 MongoDB是一个基于分布式文件存储的数据库，使用C++语言编写。旨在为Web应用提供可扩展的高性能数据存储解决方案。应用性能高低依赖于数据库性能，MongoDB则是非关系数据库中功能最丰富，最像关系数据库的，随着MongDB 3.4版本发布，其应用场景适用能力得到了进一步拓展。 MongoDB的核心优势就是...

2017-08-01

454

HBase BlockCache系列第一篇文章《走进BlockCache》从全局视角对HBase中缓存、Memstore等作了简要概述，并重点介绍了几种BlockCache方案及其演进过程，对此还不了解的可以点这里。本文在上文的基础上深入BlockCache内部，对各种BlockCache方案具体工作原理进行详细分析。Note：因为SlabCache方案在0.98版本已经不被建议使用，因此本文不针对该方案进行讲解；至于LRU方案和Bucket方案，因为后者更加复杂，本文也会花更多篇幅详细介绍该方案的实现细节。 LRUBlockCache LRUBlockCache是HBase目前默认的BlockCache机制，实现机制比较简单。它使用一个ConcurrentHashMap管理BlockKey到Block的映射关系，缓存Block只需要将BlockKey和对应的Block放入该HashMap中，查询缓存就根据BlockKey从HashMap中获取即可。同时该方案采用严格的LRU淘汰算法，当Block Cache总量达到一定阈值之后就会启动淘汰机制，最近最少使用的Block会被置换出来。在...

2017-08-01

461

资源下载

更多资源

Mario

马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长，特征是大鼻子、头戴帽子、身穿背带裤，还留着胡子。与他的双胞胎兄弟路易基一起，长年担任任天堂的招牌角色。

腾讯云软件源

为解决软件依赖安装时官方源访问速度慢的问题，腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构，目前腾讯云软件源站支持公网访问和内网访问。

Nacos

Nacos /nɑ:kəʊs/ 是 Dynamic Naming and Configuration Service 的首字母简称，一个易于构建 AI Agent 应用的动态服务发现、配置管理和AI智能体管理平台。Nacos 致力于帮助您发现、配置和管理微服务及AI智能体应用。Nacos 提供了一组简单易用的特性集，帮助您快速实现动态服务发现、服务配置、服务元数据、流量管理。Nacos 帮助您更敏捷和容易地构建、交付和管理微服务平台。

Spring

Spring框架（Spring Framework）是由Rod Johnson于2002年提出的开源Java企业级应用框架，旨在通过使用JavaBean替代传统EJB实现方式降低企业级编程开发的复杂性。该框架基于简单性、可测试性和松耦合性设计理念，提供核心容器、应用上下文、数据访问集成等模块，支持整合Hibernate、Struts等第三方框架，其适用范围不仅限于服务器端开发，绝大多数Java应用均可从中受益。