原文链接：http://www.cnblogs.com/juandx/p/4962089.html python中对文件、文件夹（文件操作函数）的操作需要涉及到os模块和shutil模块。 得到当前工作目录，即当前Python脚本工作的目录路径: os.getcwd() 返回指定目录下的所有文件和目录名:os.listdir() 函数用来删除一个文件:os.remove() 删除多个目录：os.removedirs（r“c：\python”） 检验给出的路径是否是一个文件：os.path.isfile() 检验给出的路径是否是一个目录：os.path.isdir() 判断是否是绝对路径：os.path.isabs() 检验给出的路径是否真地存:os.path.exists() 返回一个路径的目录名和文件名:os.path.split() eg os.path.split(‘/home/swaroop/byte/code/poem.txt’) 结果：(‘/home/swaroop/byte/code’, ‘poem.txt’) 分离扩展名：os.path.splitext() 获取路径名：os.path.dirname() 获取文件名：os.path.basename() 运行shell命令: os.system() 读取和设置环境变量:os.getenv() 与os.putenv() 给出当前平台使用的行终止符:os.linesep Windows使用’\r\n’，Linux使用’\n’而Mac使用’\r’ 指示你正在使用的平台：os.name 对于Windows，它是’nt’，而对于Linux/Unix用户，它是’posix’ 重命名：os.rename（old， new） 创建多级目录：os.makedirs（r“c：\python\test”） 创建单个目录：os.mkdir（“test”） 获取文件属性：os.stat（file） 修改文件权限与时间戳：os.chmod（file） 终止当前进程：os.exit（） 获取文件大小：os.path.getsize（filename） 文件操作： os.mknod(“test.txt”) 创建空文件 fp = open(“test.txt”,w) 直接打开一个文件，如果文件不存在则创建文件 关于open 模式： w 以写方式打开， a 以追加模式打开 (从 EOF 开始, 必要时创建新文件) r+ 以读写模式打开 w+ 以读写模式打开 (参见 w ) a+ 以读写模式打开 (参见 a ) rb 以二进制读模式打开 wb 以二进制写模式打开 (参见 w ) ab 以二进制追加模式打开 (参见 a ) rb+ 以二进制读写模式打开 (参见 r+ ) wb+ 以二进制读写模式打开 (参见 w+ ) ab+ 以二进制读写模式打开 (参见 a+ ) fp.read([size]) #size为读取的长度，以byte为单位 fp.readline([size]) #读一行，如果定义了size，有可能返回的只是一行的一部分 fp.readlines([size]) #把文件每一行作为一个list的一个成员，并返回这个list。其实它的内部是通过循环调用readline()来实现的。如果提供size参数，size是表示读取内容的总长，也就是说可能只读到文件的一部分。 fp.write(str) #把str写到文件中，write()并不会在str后加上一个换行符 fp.writelines(seq) #把seq的内容全部写到文件中(多行一次性写入)。这个函数也只是忠实地写入，不会在每行后面加上任何东西。 fp.close() #关闭文件。python会在一个文件不用后自动关闭文件，不过这一功能没有保证，最好还是养成自己关闭的习惯。 如果一个文件在关闭后还对其进行操作会产生ValueError fp.flush() #把缓冲区的内容写入硬盘 fp.fileno() #返回一个长整型的”文件标签“ fp.isatty() #文件是否是一个终端设备文件（unix系统中的） fp.tell() #返回文件操作标记的当前位置，以文件的开头为原点 fp.next() #返回下一行，并将文件操作标记位移到下一行。把一个file用于for … in file这样的语句时，就是调用next()函数来实现遍历的。 fp.seek(offset[,whence]) #将文件打操作标记移到offset的位置。这个offset一般是相对于文件的开头来计算的，一般为正数。但如果提供了whence参数就不一定了，whence可以为0表示从头开始计算，1表示以当前位置为原点计算。2表示以文件末尾为原点进行计算。需要注意，如果文件以a或a+的模式打开，每次进行写操作时，文件操作标记会自动返回到文件末尾。 fp.truncate([size]) #把文件裁成规定的大小，默认的是裁到当前文件操作标记的位置。如果size比文件的大小还要大，依据系统的不同可能是不改变文件，也可能是用0把文件补到相应的大小，也可能是以一些随机的内容加上去。 目录操作： os.mkdir(“file”) 创建目录 复制文件： shutil.copyfile(“oldfile”,”newfile”) oldfile和newfile都只能是文件 shutil.copy(“oldfile”,”newfile”) oldfile只能是文件夹，newfile可以是文件，也可以是目标目录 复制文件夹： shutil.copytree(“olddir”,”newdir”) olddir和newdir都只能是目录，且newdir必须不存在 重命名文件（目录） os.rename(“oldname”,”newname”) 文件或目录都是使用这条命令 移动文件（目录） shutil.move(“oldpos”,”newpos”) 删除文件 os.remove(“file”) 删除目录 os.rmdir(“dir”)只能删除空目录 shutil.rmtree(“dir”) 空目录、有内容的目录都可以删 转换目录 os.chdir(“path”) 换路径 Python读写文件 1.open 使用open打开文件后一定要记得调用文件对象的close()方法。比如可以用try/finally语句来确保最后能关闭文件。 file_object = open(‘thefile.txt’) try: all_the_text = file_object.read( ) finally: file_object.close( ) 注：不能把open语句放在try块里，因为当打开文件出现异常时，文件对象file_object无法执行close()方法。 2.读文件 读文本文件 input = open('data', 'r') #第二个参数默认为r input = open('data') 1 2 3 读二进制文件 input = open('data', 'rb') 1 读取所有内容 file_object = open('thefile.txt') try: all_the_text = file_object.read( ) finally: file_object.close( ) 1 2 3 4 5 读固定字节 file_object = open('abinfile', 'rb') try: while True: chunk = file_object.read(100) if not chunk: break do_something_with(chunk) finally: file_object.close( ) 1 2 3 4 5 6 7 8 9 读每行 list_of_all_the_lines = file_object.readlines( ) 1 如果文件是文本文件，还可以直接遍历文件对象获取每行： for line in file_object: process line 1 2 3.写文件 写文本文件 output = open('data', 'w') 1 写二进制文件 output = open('data', 'wb') 1 追加写文件 output = open('data', 'w+') 1 写数据 file_object = open('thefile.txt', 'w') file_object.write(all_the_text) file_object.close( ) 1 2 3 写入多行 file_object.writelines(list_of_text_strings) 1 注意，调用writelines写入多行在性能上会比使用write一次性写入要高。 在处理日志文件的时候，常常会遇到这样的情况：日志文件巨大，不可能一次性把整个文件读入到内存中进行处理，例如需要在一台物理内存为 2GB 的机器上处理一个 2GB 的日志文件，我们可能希望每次只处理其中 200MB 的内容。 在 Python 中，内置的 File 对象直接提供了一个 readlines(sizehint) 函数来完成这样的事情。以下面的代码为例： file = open('test.log', 'r')sizehint = 209715200 # 200Mposition = 0lines = file.readlines(sizehint)while not file.tell() - position < 0: position = file.tell() lines = file.readlines(sizehint) 1 每次调用 readlines(sizehint) 函数，会返回大约 200MB 的数据，而且所返回的必然都是完整的行数据，大多数情况下，返回的数据的字节数会稍微比 sizehint 指定的值大一点（除最后一次调用 readlines(sizehint) 函数的时候）。通常情况下，Python 会自动将用户指定的 sizehint 的值调整成内部缓存大小的整数倍。 file在python是一个特殊的类型，它用于在python程序中对外部的文件进行操作。在python中一切都是对象，file也不例外，file有file的方法和属性。下面先来看如何创建一个file对象： file(name[, mode[, buffering]]) 1 file()函数用于创建一个file对象，它有一个别名叫open()，可能更形象一些，它们是内置函数。来看看它的参数。它参数都是以字符串的形式传递的。name是文件的名字。 mode是打开的模式，可选的值为r w a U，分别代表读（默认） 写 添加支持各种换行符的模式。用w或a模式打开文件的话，如果文件不存在，那么就自动创建。此外，用w模式打开一个已经存在的文件时，原有文件的内容会被清空，因为一开始文件的操作的标记是在文件的开头的，这时候进行写操作，无疑会把原有的内容给抹掉。由于历史的原因，换行符在不同的系统中有不同模式，比如在 unix中是一个\n，而在windows中是‘\r\n’，用U模式打开文件，就是支持所有的换行模式，也就说‘\r’ ‘\n’ ‘\r\n’都可表示换行，会有一个tuple用来存贮这个文件中用到过的换行符。不过，虽说换行有多种模式，读到python中统一用\n代替。在模式字符的后面，还可以加上+ b t这两种标识，分别表示可以对文件同时进行读写操作和用二进制模式、文本模式（默认）打开文件。 buffering如果为0表示不进行缓冲;如果为1表示进行“行缓冲“;如果是一个大于1的数表示缓冲区的大小，应该是以字节为单位的。 file对象有自己的属性和方法。先来看看file的属性。 closed #标记文件是否已经关闭，由close()改写 encoding #文件编码 mode #打开模式 name #文件名 newlines #文件中用到的换行模式，是一个tuple softspace #boolean型，一般为0，据说用于print 1 2 3 4 5 6 file的读写方法： F.read([size]) #size为读取的长度，以byte为单位 F.readline([size]) #读一行，如果定义了size，有可能返回的只是一行的一部分 F.readlines([size]) #把文件每一行作为一个list的一个成员，并返回这个list。其实它的内部是通过循环调用readline()来实现的。如果提供size参数，size是表示读取内容的总长，也就是说可能只读到文件的一部分。 F.write(str) #把str写到文件中，write()并不会在str后加上一个换行符 F.writelines(seq) #把seq的内容全部写到文件中。这个函数也只是忠实地写入，不会在每行后面加上任何东西。 file的其他方法： F.close() #关闭文件。python会在一个文件不用后自动关闭文件，不过这一功能没有保证，最好还是养成自己关闭的习惯。如果一个文件在关闭后还对其进行操作会产生ValueError F.flush() #把缓冲区的内容写入硬盘 F.fileno() #返回一个长整型的”文件标签“ F.isatty() #文件是否是一个终端设备文件（unix系统中的） F.tell() #返回文件操作标记的当前位置，以文件的开头为原点 F.next() #返回下一行，并将文件操作标记位移到下一行。把一个file用于for ... in file这样的语句时，就是调用next()函数来实现遍历的。 F.seek(offset[,whence]) #将文件打操作标记移到offset的位置。这个offset一般是相对于文件的开头来计算的，一般为正数。但如果提供了whence参数就不一定了，whence可以为0表示从头开始计算，1表示以当前位置为原点计算。2表示以文件末尾为原点进行计算。需要注意，如果文件以a或a+的模式打开，每次进行写操作时，文件操作标记会自动返回到文件末尾。 F.truncate([size]) #把文件裁成规定的大小，默认的是裁到当前文件操作标记的位置。如果size比文件的大小还要大，依据系统的不同可能是不改变文件，也可能是用0把文件补到相应的大小，也可能是以一些随机的内容加上去。 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 http://www.cnblogs.com/allenblogs/archive/2010/09/13/1824842.html http://www.cnblogs.com/rollenholt/archive/2012/04/23/2466179.html 首先 dfs.replication这个参数是个client参数，即node level参数。需要在每台datanode上设置。 其实默认为3个副本已经够用了，设置太多也没什么用。 一个文件，上传到hdfs上时指定的是几个副本就是几个。以后你修改了副本数，对已经上传了的文件也不会起作用。可以再上传文件的同时指定创建的副本数 Hadoop dfs -D dfs.replication=1 -put 70M logs/2 可以通过命令来更改已经上传的文件的副本数： hadoop fs -setrep -R 3 / 查看当前hdfs的副本数 hadoop fsck -locations FSCK started by hadoop from /172.18.6.112 for path / at Thu Oct 27 13:24:25 CST 2011 ....................Status: HEALTHY Total size: 4834251860 B Total dirs: 21 Total files: 20 Total blocks (validated): 82 (avg. block size 58954290 B) Minimally replicated blocks: 82 (100.0 %) Over-replicated blocks: 0 (0.0 %) Under-replicated blocks: 0 (0.0 %) Mis-replicated blocks: 0 (0.0 %) Default replication factor: 3 Average block replication: 3.0 Corrupt blocks: 0 Missing replicas: 0 (0.0 %) Number of data-nodes: 3 Number of racks: 1 FSCK ended at Thu Oct 27 13:24:25 CST 2011 in 10 milliseconds The filesystem under path '/' is HEALTHY 某个文件的副本数，可以通过ls中的文件描述符看到 hadoop dfs -ls -rw-r--r-- 3 hadoop supergroup 153748148 2011-10-27 16:11 /user/hadoop/logs/201108/impression_witspixel2011080100.thin.log.gz 如果你只有3个datanode，但是你却指定副本数为4，是不会生效的，因为每个datanode上只能存放一个副本。 参考：http://blog.csdn.net/lskyne/article/details/8898666 转自：https://www.cnblogs.com/shabbylee/p/6792555.html 由于历史原因，Python有两个大的版本分支，Python2和Python3，又由于一些库只支持某个版本分支，所以需要在电脑上同时安装Python2和Python3，因此如何让两个版本的Python兼容，如何让脚本在对应的Python版本上运行，这个是值得总结的。 对于Ubuntu 16.04 LTS版本来说，Python2（2.7.12）和Python3（3.5.2）默认同时安装，默认的python版本是2.7.12。 当然你也可以用python2来调用。 如果想调用python3，就用python3. 对于Windows，就有点复杂了。因为不论python2还是python3，python可执行文件都叫python.exe，在cmd下输入python得到的版本号取决于环境变量里哪个版本的python路径更靠前，毕竟windows是按照顺序查找的。比如环境变量里的顺序是这样的： 那么cmd下的python版本就是2.7.12。 反之，则是python3的版本号。 这就带来一个问题了，如果你想用python2运行一个脚本，一会你又想用python3运行另一个脚本，你怎么做？来回改环境变量显然很麻烦。 网上很多办法比较简单粗暴，把两个python.exe改名啊，一个改成python2.exe，一个改成python3.exe。这样做固然可以，但修改可执行文件的方式，毕竟不是很好的方法。 我仔细查找了一些python技术文档，发现另外一个我觉得比较好的解决办法。 借用py的一个参数来调用不同版本的Python。py -2调用python2，py -3调用的是python3. 当python脚本需要python2运行时，只需在脚本前加上，然后运行py xxx.py即可。 #! python2 当python脚本需要python3运行时，只需在脚本前加上，，然后运行py xxx.py即可。 #! python3 就这么简单。 同时，这也完美解决了在pip在python2和python3共存的环境下报错，提示Fatal error in launcher: Unable to create process using '"'的问题。 当需要python2的pip时，只需 py -2 -m pip install xxx 当需要python3的pip时，只需 py -3 -m pip install xxx python2和python3的pip package就这样可以完美分开了。 Sentry权限控制通过Beeline（Hiveserver2 SQL 命令行接口）输入Grant 和 Revoke语句来配置。语法跟现在的一些主流的关系数据库很相似。需要注意的是：当sentry服务启用后，我们必须使用beeline接口来执行hive查询，Hive Cli并不支持sentry。 CREATE ROLE Statement CREATE ROLE语句创建一个可以被赋权的角色。权限可以赋给角色，然后再分配给各个用户。一个用户被分配到角色后可以执行该角色的权限。 只有拥有管理员的角色可以create/drop角色。默认情况下，hive、impala和hue用户拥有管理员角色。 CREATE ROLE [role_name]; DROP ROLE Statement DROP ROLE语句可以用来从数据库中移除一个角色。一旦移除，之前分配给所有用户的该角色将会取消。之前已经执行的语句不会受到影响。但是，因为hive在执行每条查询语句之前会检查用户的权限，处于登录活跃状态的用户会话会受到影响。 DROP ROLE [role_name]; GRANT ROLE Statement GRANT ROLE语句可以用来给组授予角色。只有sentry的管理员用户才能执行该操作。 GRANT ROLE role_name [, role_name] TO GROUP （groupName） [,GROUP （groupName）] REVOKE ROLE Statement REVOKE ROLE语句可以用来从组移除角色。只有sentry的管理员用户才能执行该操作。 REVOKE ROLE role_name [, role_name] FROM GROUP （groupName) [,GROUP (groupName)] GRANT (PRIVILEGE) Statement 授予一个对象的权限给一个角色，该用户必须为sentry的管理员用户。 GRANT (PRIVILEGE) [, (PRIVILEGE) ] ON (OBJECT) (object_name) TO ROLE (roleName) [,ROLE (roleName)] REVOKE （PRIVILEGE） Statement 因为只有认证的管理员用户可以创建角色，从而只有管理员用户可以取消一个组的权限。 REVOKE (PRIVILEGE) [, (PRIVILEGE) ] ON (OBJECT) (object_name) FROM ROLE (roleName) [,ROLE (roleName)] GRANT (PRIVILEGE) ... WITH GRANT OPTION 在cdh5.2中，你可以委托给其他角色来授予和解除权限。比如，一个角色被授予了WITH GRANT OPTION的权限可以GRANT/REVOKE同样的权限给其他角色。因此，如果一个角色有一个库的所有权限并且设置了 WITH GRANT OPTION，该角色分配的用户可以对该数据库和其中的表执行GRANT/REVOKE语句。 GRANT （PRIVILEGE） ON （OBJECT） （object_name） TO ROLE （roleName） WITH GRANT OPTION 只有一个带GRANT选项的特殊权限的角色或者它的父级权限可以从其他角色解除这种权限。一旦下面的语句执行，所有跟其相关的grant权限将会被解除。 REVOKE （RIVILEGE） ON （BJECT） （bject_name） FROM ROLE （roleName） Hive目前不支持解除之前赋予一个角色 WITH GRANT OPTION 的权限。要想移除WITH GRANT OPTION、解除权限，可以重新去除 WITH GRANT OPTION这个标记来再次附权。 SET ROLE Statement SET ROLE语句可以给当前会话选择一个角色使之生效。一个用户只能启用分配给他的角色。任何不存在的角色和当前用户不能使用的角色是不能生效的。如果没有使用任何角色，用户将会使用任何一个属于他的角色的权限。 选择一个角色使用： To enable a specific role: 使用所有的角色： To enable a specific role: 关闭所有角色 SET ROLE NONE; SHOW Statement 显示当前用户拥有库、表、列相关权限的数据库： SHOW DATABASES; 显示当前用户拥有表、列相关权限的表； SHOW TABLES; 显示当前用户拥有SELECT权限的列： SHOW COLUMNS (FROM|IN) table_name [(FROM|IN) db_name]; 显示当前系统中所有的角色（只有管理员用户可以执行）： SHOW ROLES; 显示当前影响当前会话的角色： SHOW CURRENT ROLES; 显示指定组的被分配到的所有角色（只有管理员用户和指定组内的用户可以执行） SHOW ROLE GRANT GROUP (groupName); SHOW语句可以用来显示一个角色被授予的权限或者显示角色的一个特定对象的所有权限。 显示指定角色的所有被赋予的权限。（只有管理员用户和指定角色分配到的用户可以执行）。下面的语句也会显示任何列级的权限。 SHOW GRANT ROLE (roleName); 显示指定对象的一个角色的所有被赋予的权限（只有管理员用户和指定角色分配到的用户可以执行）。下面的语句也会显示任何列级的权限。 SHOW GRANT ROLE (roleName) on (OBJECT) (objectName); ----------------------------我也是有底线的-----------------------------      摘要: Python 里面的编码和解码也就是 unicode 和 str 这两种形式的相互转化。编码是 unicode -> str，相反的，解码就是 str -> unicode。剩下的问题就是确定何时需要进行编码或者解码了.关于文件开头的"编码指示"，也就是 # -*- codin...  阅读全文      摘要: 我们每次执行hive的hql时，shell里都会提示一段话：[python] view plaincopy...  Number of reduce tasks not specified. Estimated from input data size: 50...  阅读全文      摘要: spark 累加历史主要用到了窗口函数，而进行全部统计，则需要用到rollup函数 1  应用场景： 　　1、我们需要统计用户的总使用时长（累加历史） 　　2、前台展现页面需要对多个维度进行查询，如：产品、地区等等 　　3、需要展现的表格头如： 产品、2015-04、2015-05、2015-06 2 原始数据： product_code |event_date |dur...  阅读全文      摘要: Spark1.4发布，支持了窗口分析函数(window functions)。在离线平台中，90%以上的离线分析任务都是使用Hive实现，其中必然会使用很多窗口分析函数，如果SparkSQL支持窗口分析函数， 那么对于后面Hive向SparkSQL中的迁移的工作量会大大降低，使用方式如下： 1、初始化数据 创建表 [sql] view plain cop...  阅读全文

SparkSQL相关语句总结

如果用传统SCP远程拷贝，速度是比较慢的。现在采用lz4压缩传输。LZ4是一个非常快的无损压缩算法，压缩速度在单核300MB/S，可扩展支持多核CPU。它还具有一个非常快速的解码器，速度单核可达到和超越1GB/S。通常能够达到多核系统上的RAM速度限制。 你PV 全命为Pipe Viewer，利用它我们可以查看到命令执行的进度。 下面介绍下lz4和pv的安装，下载软件： 下载pv-1.1.4.tar.gz wget http://sourceforge.jp/projects/sfnet_pipeviewer/downloads/pipeviewer/1.1.4/pv-1.1.4.tar.bz2/ 下lz4的包难一些，可能要FQ：https://dl.dropboxusercontent.com/u/59565338/LZ4/lz4-r108.tar.gz 安装灰常简单： pv安装： [root ~]$ tar jxvf pv-1.1.4.tar.bz2 [root ~]$ cd pv-1.1.4 [root pv-1.1.4]$ ./configure && make && make install lz4安装： [root ~]$ tar zxvf lz4-r108.tar.gz [root ~]$ cd lz4-r108 [root lz4-r108]$ make && make install 用法：（-c 后指定要传输的文件，ssh -p 是指定端口，后面的ip是目标主机的ip, -xC指定传到目标主机下的那个目录下，别的不用修改）： tar -c mysql-slave-3307 |pv|lz4 -B4|ssh -p10022 -c arcfour128 -o"MACs umac-64@openssh.com" 192.168.100.234 "lz4 -d |tar -xC /data" 下面是我线上传一个从库的效果： 看到了吧，25.7G 只需要接近3分钟，这样远比scp速度快上了好几倍，直接scp拷贝离散文件，很消耗IO，而使用LZ4快速压缩，对性能影响不大，传输速度快 PS：下次补充同机房不同网段的传输效果及跨机房的传输效果^0^ 作者：陆炫志 出处：xuanzhi的博客 http://www.cnblogs.com/xuanzhi201111 您的支持是对博主最大的鼓励，感谢您的认真阅读。本文版权归作者所有，欢迎转载，但请保留该声明。 王 腾腾 和 邵 兵 2015 年 11 月 26 日发布 WeiboGoogle+用电子邮件发送本页面 Comments 1 引子 随着云时代的来临，大数据（Big data）也获得了越来越多的关注。著云台的分析师团队认为，大数据（Big data）通常用来形容一个公司创造的大量非结构化和半结构化数据，这些数据在下载到关系型数据库用于分析时会花费过多时间和金钱。大数据分析常和云计算联系到一起，因为实时的大型数据集分析需要像 MapReduce 一样的框架来向数十、数百或甚至数千的电脑分配工作。 “大数据”在互联网行业指的是这样一种现象：互联网公司在日常运营中生成、累积的用户网络行为数据。这些数据的规模是如此庞大，以至于不能用 G 或 T 来衡量。所以如何高效的处理分析大数据的问题摆在了面前。对于大数据的处理优化方式有很多种，本文中主要介绍在使用 Hadoop 平台中对数据进行压缩处理来提高数据处理效率。 压缩简介 Hadoop 作为一个较通用的海量数据处理平台，每次运算都会需要处理大量数据，我们会在 Hadoop 系统中对数据进行压缩处理来优化磁盘使用率，提高数据在磁盘和网络中的传输速度，从而提高系统处理数据的效率。在使用压缩方式方面，主要考虑压缩速度和压缩文件的可分割性。综合所述，使用压缩的优点如下： 1. 节省数据占用的磁盘空间； 2. 加快数据在磁盘和网络中的传输速度，从而提高系统的处理速度。 压缩格式 Hadoop 对于压缩格式的是自动识别。如果我们压缩的文件有相应压缩格式的扩展名（比如 lzo，gz，bzip2 等）。Hadoop 会根据压缩格式的扩展名自动选择相对应的解码器来解压数据，此过程完全是 Hadoop 自动处理，我们只需要确保输入的压缩文件有扩展名。 Hadoop 对每个压缩格式的支持, 详细见下表： 表 1. 压缩格式 压缩格式 工具 算法 扩展名 多文件 可分割性 DEFLATE 无 DEFLATE .deflate 不 不 GZIP gzip DEFLATE .gzp 不 不 ZIP zip DEFLATE .zip 是 是，在文件范围内 BZIP2 bzip2 BZIP2 .bz2 不 是 LZO lzop LZO .lzo 不 是 如果压缩的文件没有扩展名，则需要在执行 MapReduce 任务的时候指定输入格式。 1 2 3 4 5 hadoop jar /usr/home/hadoop/hadoop-0.20.2/contrib/streaming/ hadoop-streaming-0.20.2-CD H3B4.jar -file /usr/home/hadoop/hello/mapper.py -mapper / usr/home/hadoop/hello/mapper.py -file /usr/home/hadoop/hello/ reducer.py -reducer /usr/home/hadoop/hello/reducer.py -input lzotest -output result4 - jobconf mapred.reduce.tasks=1*-inputformatorg.apache.hadoop.mapred.LzoTextInputFormat* 性能对比 Hadoop 下各种压缩算法的压缩比，压缩时间，解压时间见下表: 表 2. 性能对比 压缩算法 原始文件大小 压缩文件大小 压缩速度 解压速度 gzip 8.3GB 1.8GB 17.5MB/s 58MB/s bzip2 8.3GB 1.1GB 2.4MB/s 9.5MB/s LZO-bset 8.3GB 2GB 4MB/s 60.6MB/s LZO 8.3GB 2.9GB 49.3MB/s 74.6MB/s 因此我们可以得出： 1) Bzip2 压缩效果明显是最好的，但是 bzip2 压缩速度慢，可分割。 2) Gzip 压缩效果不如 Bzip2，但是压缩解压速度快，不支持分割。 3) LZO 压缩效果不如 Bzip2 和 Gzip，但是压缩解压速度最快！并且支持分割！ 这里提一下，文件的可分割性在 Hadoop 中是很非常重要的，它会影响到在执行作业时 Map 启动的个数，从而会影响到作业的执行效率！ 所有的压缩算法都显示出一种时间空间的权衡，更快的压缩和解压速度通常会耗费更多的空间。在选择使用哪种压缩格式时，我们应该根据自身的业务需求来选择。 下图是在本地压缩与通过流将压缩结果上传到 BI 的时间对比。 图 1. 时间对比 图 1. 时间对比 使用方式 MapReduce 可以在三个阶段中使用压缩。 1. 输入压缩文件。如果输入的文件是压缩过的，那么在被 MapReduce 读取时，它们会被自动解压。 2.MapReduce 作业中，对 Map 输出的中间结果集压缩。实现方式如下： 1）可以在 core-site.xml 文件中配置，代码如下 图 2. core-site.xml 代码示例 图 2. core-site.xml 代码示例 2）使用 Java 代码指定 1 2 conf.setCompressMapOut(true); conf.setMapOutputCompressorClass(GzipCode.class); 最后一行代码指定 Map 输出结果的编码器。 3.MapReduce 作业中，对 Reduce 输出的最终结果集压。实现方式如下： 1）可以在 core-site.xml 文件中配置，代码如下 图 3. core-site.xml 代码示例 图 3. core-site.xml 代码示例 2）使用 Java 代码指定 1 2 conf.setBoolean(“mapred.output.compress”,true); conf.setClass(“mapred.output.compression.codec”,GzipCode.class,CompressionCodec.class); 最后一行同样指定 Reduce 输出结果的编码器。 压缩框架 我们前面已经提到过关于压缩的使用方式，其中第一种就是将压缩文件直接作为入口参数交给 MapReduce 处理，MapReduce 会自动根据压缩文件的扩展名来自动选择合适解压器处理数据。那么到底是怎么实现的呢？如下图所示： 图 4. 压缩实现情形 图 4. 压缩实现情形 我们在配置 Job 作业的时候，会设置数据输入的格式化方式，使用 conf.setInputFormat() 方法，这里的入口参数是 TextInputFormat.class。 TextInputFormat.class 继承于 InputFormat.class，主要用于对数据进行两方面的预处理。一是对输入数据进行切分，生成一组 split，一个 split 会分发给一个 mapper 进行处理；二是针对每个 split，再创建一个 RecordReader 读取 split 内的数据，并按照的形式组织成一条 record 传给 map 函数进行处理。此类在对数据进行切分之前，会首先初始化压缩解压工程类 CompressionCodeFactory.class，通过工厂获取实例化的编码解码器 CompressionCodec 后对数据处理操作。 下面我们来详细的看一下从压缩工厂获取编码解码器的过程。 压缩解压工厂类 CompressionCodecFactory 压缩解压工厂类 CompressionCodeFactory.class 主要功能就是负责根据不同的文件扩展名来自动获取相对应的压缩解压器 CompressionCodec.class，是整个压缩框架的核心控制器。我们来看下 CompressionCodeFactory.class 中的几个重要方法： 1. 初始化方法 图 5. 代码示例 图 5. 代码示例 ① getCodeClasses(conf) 负责获取关于编码解码器 CompressionCodec.class 的配置信息。下面将会详细讲解。 ② 默认添加两种编码解码器。当 getCodeClass(conf) 方法没有读取到相关的编码解码器 CompressionCodec.class 的配置信息时，系统会默认添加两种编码解码器 CompressionCodec.class，分别是 GzipCode.class 和 DefaultCode.class。 ③ addCode(code) 此方法用于将编码解码器 CompressionCodec.class 添加到系统缓存中。下面将会详细讲解。 2. getCodeClasses(conf) 图 6. 代码示例 图 6. 代码示例 ① 这里我们可以看，系统读取关于编码解码器 CompressionCodec.class 的配置信息在 core-site.xml 中 io.compression.codes 下。我们看下这段配置文件，如下图所示： 图 7. 代码示例 图 7. 代码示例 Value 标签中是每个编码解码 CompressionCodec.class 的完整路径，中间用逗号分隔。我们只需要将自己需要使用到的编码解码配置到此属性中，系统就会自动加载到缓存中。 除了上述的这种方式以外，Hadoop 为我们提供了另一种加载方式：代码加载。同样最终将信息配置在 io.compression.codes 属性中，代码如下： 1 2 conf.set("io.compression.codecs","org.apache.hadoop.io.compress.DefaultCodec, org.apache.hadoop.io.compress.GzipCodec,com.hadoop.compression.lzo.LzopCodec");) 3. addCode(code) 方法添加编码解码器 图 8. 代码示例 图 8. 代码示例 addCodec(codec) 方法入口参数是个编码解码器 CompressionCodec.class，这里我们会首先接触到它的一个方法。 ① codec.getDefaultExtension() 方法看方法名的字面意思我们就可以知道，此方法用于获取此编码解码所对应文件的扩展名，比如，文件名是 xxxx.gz2，那么这个方法的返回值就是“.bz2”，我们来看下 org.apache.hadoop.io.compress.BZip2Codec 此方法的实现代码： 图 9. 代码示例 图 9. 代码示例 ② Codecs 是一个 SortedMap 的示例。这里有个很有意思的地方，它将 Key 值，也就是通过 codec.getDefaultExtension() 方法获取到的文件扩展名进行了翻转，举个例子，比如文件名扩展名“.bz2”，将文件名翻转之后就变成了“2zb.”。 系统加载完所有的编码解码器后，我们可以得到这样一个有序映射表，如下： 图 10. 代码示例 图 10. 代码示例 现在编码解码器都有了，我们怎么得到对应的编码解码器呢？看下面这个方法。 4. getCodec() 方法 此方法用于获取文件所对应的的编码解码器 CompressionCodec.class。 图 11. 代码示例 图 11. 代码示例 getCodec(Path) 方法的输入参数是 Path 对象，保存着文件路径。 ① 将文件名翻转。如 xxxx.bz2 翻转成 2zb.xxxx。 ② 获取 codecs 集合中最接近 2zb.xxxx 的值。此方法有返回值同样是个 SortMap 对象。 在这里对返回的 SortMap 对象进行第二次筛选。 编码解码器 CompressionCodec 刚刚在介绍压缩解压工程类 CompressionCodeFactory.class 的时候，我们多次提到了压缩解压器 CompressionCodecclass，并且我们在上文中还提到了它其中的一个用于获取文件扩展名的方法 getDefaultExtension()。 压缩解压工程类 CompressionCodeFactory.class 使用的是抽象工厂的设计模式。它是一个接口，制定了一系列方法，用于创建特定压缩解压算法。下面我们来看下比较重要的几个方法： 1. createOutputStream() 方法对数据流进行压缩。 图 12. 代码示例 图 12. 代码示例 此方法提供了方法重载。 ① 基于流的压缩处理； ② 基于压缩机 Compress.class 的压缩处理 2. createInputStream() 方法对数据流进行解压。 图 13. 代码示例 图 13. 代码示例 这里的解压方法同样提供了方法重载。 ① 基于流的解压处理； ② 基于解压机 Decompressor.class 的解压处理； 关于压缩/解压流与压缩/解压机会在下面的文章中我们会详细讲解。此处暂作了解。 3. getCompressorType() 返回需要的编码器的类型。 getDefaultExtension() 获取对应文件扩展名的方法。前文已提到过，不再敖述。 压缩机 Compressor 和解压机 Decompressor 前面在编码解码器部分的 createInputStream() 和 createInputStream() 方法中我们提到过 Compressor.class 和 Decompressor.class 对象。在 Hadoop 的实现中，数据编码器和解码器被抽象成了两个接口： 1. org.apache.hadoop.io.compress.Compressor; 2. org.apache.hadoop.io.compress.Decompressor; 它们规定了一系列的方法，所以在 Hadoop 内部的编码/解码算法实现都需要实现对应的接口。在实际的数据压缩与解压缩过程，Hadoop 为用户提供了统一的 I/O 流处理模式。 我们看一下压缩机 Compressor.class，代码如下： 图 14. 代码示例 图 14. 代码示例 ① setInput() 方法接收数据到内部缓冲区，可以多次调用； ② needsInput() 方法用于检查缓冲区是否已满。如果是 false 则说明当前的缓冲区已满； ③ getBytesRead() 输入未压缩字节的总数； ④ getBytesWritten() 输出压缩字节的总数； ⑤ finish() 方法结束数据输入的过程； ⑥ finished() 方法用于检查是否已经读取完所有的等待压缩的数据。如果返回 false，表明压缩器中还有未读取的压缩数据，可以继续通过 compress() 方法读取； ⑦ compress() 方法获取压缩后的数据，释放缓冲区空间； ⑧ reset() 方法用于重置压缩器，以处理新的输入数据集合； ⑨ end() 方法用于关闭解压缩器并放弃所有未处理的输入； ⑩ reinit() 方法更进一步允许使用 Hadoop 的配置系统，重置并重新配置压缩器； 为了提高压缩效率，并不是每次用户调用 setInput() 方法，压缩机就会立即工作，所以，为了通知压缩机所有数据已经写入，必须使用 finish() 方法。finish() 调用结束后，压缩机缓冲区中保持的已经压缩的数据，可以继续通过 compress() 方法获得。至于要判断压缩机中是否还有未读取的压缩数据，则需要利用 finished() 方法来判断。 压缩流 CompressionOutputStream 和解压缩流 CompressionInputStream 前文编码解码器部分提到过 createInputStream() 方法返回 CompressionOutputStream 对象，createInputStream() 方法返回 CompressionInputStream 对象。这两个类分别继承自 java.io.OutputStream 和 java.io.InputStream。从而我们不难理解，这两个对象的作用了吧。 我们来看下 CompressionInputStream.class 的代码： 图 15. 代码示例 图 15. 代码示例 可以看到 CompressionOutputStream 实现了 OutputStream 的 close() 方法和 flush() 方法，但用于输出数据的 write() 方法以及用于结束压缩过程并将输入写到底层流的 finish() 方法和重置压缩状态的 resetState() 方法还是抽象方法，需要 CompressionOutputStream 的子类实现。 Hadoop 压缩框架中为我们提供了一个实现了 CompressionOutputStream 类通用的子类 CompressorStream.class。 图 16. 代码示例 图 16. 代码示例 CompressorStream.class 提供了三个不同的构造函数，CompressorStream 需要的底层输出流 out 和压缩时使用的压缩器，都作为参数传入构造函数。另一个参数是 CompressorStream 工作时使用的缓冲区 buffer 的大小，构造时会利用这个参数分配该缓冲区。第一个可以手动设置缓冲区大小，第二个默认 512，第三个没有缓冲区且不可使用压缩器。 图 17. 代码示例 图 17. 代码示例 在 write()、compress()、finish() 以及 resetState() 方法中，我们发现了压缩机 Compressor 的身影，前面文章我们已经介绍过压缩机的的实现过程，通过调用 setInput() 方法将待压缩数据填充到内部缓冲区，然后调用 needsInput() 方法检查缓冲区是否已满，如果缓冲区已满，将调用 compress() 方法对数据进行压缩。流程如下图所示： 图 18. 调用流程图 图 18. 调用流程图 结束语 本文深入到 Hadoop 平台压缩框架内部，对其核心代码以及各压缩格式的效率进行对比分析，以帮助读者在使用 Hadoop 平台时，可以通过对数据进行压缩处理来提高数据处理效率。当再次面临海量数据处理时， Hadoop 平台的压缩机制可以让我们事半功倍。 相关主题 Hadoop 在线 API 《Hadoop 技术内幕深入解析 HADOOP COMMON 和 HDFS 架构设计与实现原理》 developerWorks 开源技术主题：查找丰富的操作信息、工具和项目更新，帮助您掌握开源技术并将其用于 IBM 产品。

posted @ 2017-09-14 17:35 xzc 阅读(554) | 评论 (0) | 编辑 收藏

Linux系统查看当前主机CPU、内存、机器型号及主板信息

Linux系统查看当前主机CPU、内存、机器型号及主板信息：

查看CPU信息（型号）
# cat /proc/cpuinfo | grep name | cut -f2 -d: | uniq -c

 

查看内存信息
# cat /proc/meminfo

 

查看主板型号：
# dmidecode |grep -A16 "System Information$"

 

查看机器型号
# dmidecode | grep "Product Name"

 

查看当前操作系统内核信息
# uname -a

 

查看当前操作系统发行版信息
# cat /etc/issue | grep Linux

posted @ 2017-09-10 16:37 xzc 阅读(239) | 评论 (0) | 编辑 收藏

Hadoop YARN最近几个新特性

本文介绍Hadoop YARN最近版本中增加的几个非常有用的特性，包括： （1）ResourceManager HA 在apache hadoop 2.4或者CDH5.0.0版本之后，增加了ResourceManger HA特性，支持基于Zookeeper的热主备切换，具体配置参数可以参考Cloudera的文档：ResourceManager HA配置。 需要注意的是，ResourceManager HA只完成了第一个阶段的设计，即备ResourceManager启动后，会杀死之前正在运行的Application，然后从共享存储系统中读取这些Application的元数据信息，并重新提交这些Application。启动ApplicationMaster后，剩下的容错功能就交给ApplicationMaster实现了，比如MapReduce的ApplicationMaster会不断地将完成的任务信息写到HDFS上，这样，当它重启时，可以重新读取这些日志，进而只需重新运行那些未完成的任务。ResourceManager HA第二个阶段的任务是，备ResourceManager接管主ResourceManager后，无需杀死那些正在运行的Application，让他们像任何事情没有发生一样运行下去。 （2） 磁盘容错 在apache hadoop 2.4或者CDH5.0.0版本之后，增加了几个对多磁盘非常友好地参数，这些参数允许YARN更好地使用NodeManager上的多块磁盘，相关jira为：YARN-1781，主要新增了三个参数： yarn.nodemanager.disk-health-checker.min-healthy-disks：NodeManager上最少保证健康磁盘比例，当健康磁盘比例低于该值时，NodeManager不会再接收和启动新的Container，默认值是0.25，表示25%； yarn.nodemanager.disk-health-checker.max-disk-utilization-per-disk-percentage：一块磁盘的最高使用率，当一块磁盘的使用率超过该值时，则认为该盘为坏盘，不再使用该盘，默认是100，表示100%，可以适当调低； yarn.nodemanager.disk-health-checker.min-free-space-per-disk-mb：一块磁盘最少保证剩余空间大小，当某块磁盘剩余空间低于该值时，将不再使用该盘，默认是0，表示0MB。 （3）资源调度器 Fair Scheduler：Fair Scheduler增加了一个非常有用的新特性，允许用户在线将一个应用程序从一个队列转移到另外一个队列，比如将一个重要作业从一个低优先级队列转移到高优先级队列，操作命令是：bin/yarn application -movetoqueue appID -queue targetQueueName，相关jira为：YARN-1721。 Capacity Scheduler：Capacity Scheduler中资源抢占功能经过了充分的测试，可以使用了。 原创文章，转载请注明： 转载自董的博客 本文链接地址: http://dongxicheng.org/mapreduce-nextgen/hadoop-yarn-recently-new-features/

posted @ 2017-09-07 11:37 xzc 阅读(271) | 评论 (0) | 编辑 收藏

apreduce on yarn简单内存分配解释

关于mapreduce程序运行在yarn上时内存的分配一直是一个让我蒙圈的事情，单独查任何一个资料都不能很好的理解透彻。于是，最近查了大量的资料，综合各种解释，终于理解到了一个比较清晰的程度，在这里将理解的东西做一个简单的记录，以备忘却。 首先，先将关于mapreduce和yarn关于内存分配的参数粘贴上: yarn.scheduler.minimum-allocation-mb yarn.scheduler.maximum-allocation-mb yarn.nodemanager.resource.memory-mb yarn.nodemanager.vmem-pmem-ratio yarn.scheduler.increment-allocation-mb mapreduce.map.memory.mb mapreduce.reduce.memory.mb mapreduce.map.java.opts mapreduce.reduce.java.opts 个人认为，针对mapreduce任务，这些参数只有放在一起学习才能真正理解，如果单独考虑，理解不清晰。下面开始详细讲解。 一、理解参数yarn.nodemanager.resource.memory-mb,yarn.nodemanager.vmem-pmem-ratio yarn.nodemanager.resource.memory-mb很简单，就是你的这台服务器节点上准备分给yarn的内存; yarn.nodemanager.vmem-pmem-ratio网上解释都是"每使用1MB物理内存，最多可用的虚拟内存数，默认2.1"，但是目前我还是不太理解其作用是什么，有知道的朋友希望能详细解释下。 二、理解参数yarn.scheduler.minimum-allocation-mb和yarn.scheduler.maximum-allocation-mb 都知道，在yarn上运行程序时每个task都是在独立的Container中运行的，单个Container可以申请的最小和最大内存的限制就是这两个参数，注意，并不是这两个参数决定单个Container申请内存的大小，而仅仅是限制的一个范围。 三、理解yarn的内存规整化因子和内存规整化算法 先不说和哪个参数有关，单纯理解这一概念。举例: 假如规整化因子b=512M，上述讲的参数yarn.scheduler.minimum-allocation-mb为1024，yarn.scheduler.maximum-allocation-mb为8096，然后我打算给单个map任务申请内存资源(mapreduce.map.memory.mb): 申请的资源为a=1000M时，实际得到的Container内存大小为1024M(小于yarn.scheduler.minimum-allocation-mb的话自动设置为yarn.scheduler.minimum-allocation-mb); 申请的资源为a=1500M时，实际得到的Container内存大小为1536M，计算公式为:ceiling(a/b)*b，即ceiling(a/b)=ceiling(1500/512)=3,3*512=1536。此处假如b=1024，则Container实际内存大小为2048M 也就是说Container实际内存大小最小为yarn.scheduler.minimum-allocation-mb值，然后增加时的最小增加量为规整化因子b，最大不超过yarn.scheduler.maximum-allocation-mb 四、理解mapreduce.map.memory.mb、mapreduce.reduce.memory.mb "三"中提到的"打算给单个map任务申请内存资源"也就是a,其实就是指的"mapreduce.map.memory.mb"或"mapreduce.reduce.memory.mb"，注意其值不要超过yarn.scheduler.maximum-allocation-mb 五、理解mapreduce.map.java.opts、mapreduce.reduce.java.opts 以map任务为例，Container其实就是在执行一个脚本文件，而脚本文件中，会执行一个 Java 的子进程，这个子进程就是真正的 Map Task，mapreduce.map.java.opts 其实就是启动 JVM 虚拟机时，传递给虚拟机的启动参数，而默认值 -Xmx200m 表示这个 Java 程序可以使用的最大堆内存数，一旦超过这个大小，JVM 就会抛出 Out of Memory 异常，并终止进程。而 mapreduce.map.memory.mb 设置的是 Container 的内存上限，这个参数由 NodeManager 读取并进行控制，当 Container 的内存大小超过了这个参数值，NodeManager 会负责 kill 掉 Container。在后面分析 yarn.nodemanager.vmem-pmem-ratio 这个参数的时候，会讲解 NodeManager 监控 Container 内存（包括虚拟内存和物理内存）及 kill 掉 Container 的过程。 也就是说，mapreduce.map.java.opts一定要小于mapreduce.map.memory.mb mapreduce.reduce.java.opts同mapreduce.map.java.opts一样的道理。 六、理解规整化因子指的是哪个参数 "三"中提到的规整化因子也就是b，具体指的是哪个参数和yarn使用的调度器有关，一共有三种调度器:capacity scheduler（默认调度器）、fair scheduler和fifo scheduler 当使用capacity scheduler或者fifo scheduler时，规整化因子指的就是参数yarn.scheduler.minimum-allocation-mb，不能单独配置，即yarn.scheduler.increment-allocation-mb无作用; 当使用fair scheduler时，规整化因子指的是参数yarn.scheduler.increment-allocation-mb 至此，关于yarn和mapreduce的任务内存配置问题讲完了，这也是我目前理解的层次。

posted @ 2017-08-30 21:05 xzc 阅读(304) | 评论 (0) | 编辑 收藏