Tachyon的配置和使用入门

原创: 2016-06-07 16:41:09 1068浏览

Tachyon是一个基于内存的分布式文件系统（项目首页：tachyon-project.org），它是AmpLab的BDAS(berkeley data analytics stack)的一个重要组成。解决了丢失cache导致的重新计算，不同app（job），甚至是不同计算框架间重复的内存使用等问题。目前Spark 1.1默

特性

Java-like File API: Tachyon的原生API和Java文件系统非常相似，提供InputStream, OutputStream等接口, 以及高效的内存映射I/O，用这些API能够获得最好的性能。
Compatibility: Tachyon 实现了Hadoop FileSystem 接口, 因此Hadoop MapReduce和Spark可以不经过任何修改就能在使用Tachyon。
Native support for raw tables: Tachyon对列存储结构的数据提供了原生的支持，用户可以将某些访问量高的列选择性地放到内存中。
Pluggable underlayer file system: Tachyon 提供memory data到底层文件系统的方法。目前支持HDFS和单点的本地文件系统。
Web UI: 用户可以通过浏览器浏览文件系统，在debug模式下，管理员可以查看文件的位置等详细信息。
Command line interaction: 用户可以使用 ./bin/tachyon tfs和 Tachyon交互，例如将文件在Tachyon和本地文件系统中拷贝。

原理简述

参考Dr.浩源 6月30日的slide。Tachyon的架构是常见的Master/Worker结构，使用Zookeeper可以构建Master的HA。由Master节点负责管理维护文件系统MetaData（使用Journal image+edit log，详见参考1)，而文件数据维护在Worker节点的内存中。Worker和Master的通讯依赖于thrift。另外，底层支持用户指定文件的持久化（保存到underlyHDFS中）。
tachyon-arch
Tachyon充分利用内存，在内存中只存一份数据（没有replica复制内存数据），并将lineage的设计应用到存储层，通过异步的向Tachyon的底层文件系统做Checkpoint。当我们向Tachyon里面写入文件的时候，Tachyon会在后台异步的把这个文件给checkpoint到它的底层存储。另外，Tachyon的重算如下图，如果File Set B丢失，则需要由File Set A通过Spark Job重新得到File Set B。
tachyon-recompute
Tachyon中定义了下面几种cache的类型

package tachyon.client
import java.io.IOException;
/**
 * Different write types for a TachyonFile.
 */
public enum WriteType {
  /**
   * Write the file and must cache it.
   */
  MUST_CACHE(1),
  /**
   * Write the file and try to cache it.
   */
  TRY_CACHE(2),
  /**
   * Write the file synchronously to the under fs, and also try to cache it,
   */
  CACHE_THROUGH(3),
  /**
   * Write the file synchronously to the under fs, no cache.
   */
  THROUGH(4),
  /**
   * Write the file asynchronously to the under fs (either must cache or must through).
   */
  ASYNC_THROUGH(5);
......

Tachyon集群配置

下载并解压Tachyon 0.5
wget http://tachyon-project.org/downloads/tachyon-0.5.0-bin.tar.gz
tar xvfz tachyon-0.5.0-bin.tar.gz
cd tachyon-0.5.0/conf
Tachyon官方文档Configuration Settings，除了设置正确的JAVA_HOME，我们要设置的参数如下：

#Basic
tachyon.home = /var/lib/spark/tachyon-0.5.0
tachyon.underfs.address = hdfs://hdp01:8020
tachyon.data.folder = /user/spark/tach_data
tachyon.workers.folder = /user/spark/tach_worker
# tachyon.underfs.hdfs.impl = "org.apache.hadoop.hdfs.DistributedFileSystem" #default
# tachyon.max.columns = 1000 #default
# tachyon.table.metadata.byte = 5242880 #default
#HA
tachyon.usezookeeper = true
tachyon.zookeeper.address = hdp02:2181, hdp03:2181, hdp04:2181
tachyon.zookeeper.election.path = "/tach_elect"
tachyon.zookeeper.leader.path = "/tach_leader"
#Master
# tachyon.master.journal.folder = "$TACHYON_UNDERFS_ADDRESS/user/spark/tach_journal/"  #default $tachyon.home + "/journal/"    
tachyon.master.hostname = hdp04
# tachyon.master.port = 19998    #default 
# tachyon.master.web.port = 19999    #default 
# tachyon.master.whitelist = "//m.sbmmt.com/m/"    #default 
#Worker
# tachyon.worker.port = 29998 #default 
# tachyon.worker.data.port = 29999 #default 
tachyon.worker.memory.size = 10G      #default 128M
tachyon.worker.data.folder = /mnt/ramdisk           #default /mnt/ramdisk
#User
# tachyon.user.failed.space.request.limits = 3    #default 
# tachyon.user.quota.unit.bytes = 8MB    #default 
# tachyon.user.file.buffer.bytes = 1MB    #default 
# tachyon.user.default.block.size.byte = 1GB    #default 
# tachyon.user.remote.read.buffer.size.byte = 1MB    #default

如果启用了基于ZooKeeper的master HA，除了设置underfs为分布式文件系统和zk之外，还需要设置所有master的tachyon.master.hostname为自身的地址（必须对所有worker节点可见）。根据模版，配置tachyon-env.sh.template文件,tachyon-env.sh的如下：

#!/usr/bin/env bash
# This file contains environment variables required to run Tachyon. Copy it as tachyon-env.sh and
# edit that to configure Tachyon for your site. At a minimum,
# the following variables should be set:
#
# - JAVA_HOME, to point to your JAVA installation
# - TACHYON_MASTER_ADDRESS, to bind the master to a different IP address or hostname
# - TACHYON_UNDERFS_ADDRESS, to set the under filesystem address.
# - TACHYON_WORKER_MEMORY_SIZE, to set how much memory to use (e.g. 1000mb, 2gb) per worker
# - TACHYON_RAM_FOLDER, to set where worker stores in memory data
# - TACHYON_UNDERFS_HDFS_IMPL, to set which HDFS implementation to use (e.g. com.mapr.fs.MapRFileSystem,
#   org.apache.hadoop.hdfs.DistributedFileSystem)
# The following gives an example:
export TACHYON_HOME=/var/lib/spark/tachyon-0.5.0
export HADOOP_HOME=/usr/lib/hadoop
if [[ `uname -a` == Darwin* ]]; then
  # Assuming Mac OS X
  export JAVA_HOME=${JAVA_HOME:-$(/usr/libexec/java_home)}
  export TACHYON_RAM_FOLDER=/var/lib/spark/tachyon-0.5.0
  export TACHYON_JAVA_OPTS="-Djava.security.krb5.realm= -Djava.security.krb5.kdc="
else
  # Assuming Linux
  if [ -z "$JAVA_HOME" ]; then
    export JAVA_HOME=/usr/java/latest
 fi
  export TACHYON_RAM_FOLDER=$TACHYON_HOME/ramdisk
fi
export JAVA="$JAVA_HOME/bin/java"
export TACHYON_MASTER_ADDRESS=hdp04
export TACHYON_UNDERFS_ADDRESS=hdfs://hdp01:8020
export TACHYON_WORKER_MEMORY_SIZE=10GB
export TACHYON_UNDERFS_HDFS_IMPL=org.apache.hadoop.hdfs.DistributedFileSystem
CONF_DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )"
export TACHYON_JAVA_OPTS+="
  -Dlog4j.configuration=file:$CONF_DIR/log4j.properties
  -Dtachyon.debug=false
  -Dtachyon.underfs.address=$TACHYON_UNDERFS_ADDRESS
  -Dtachyon.underfs.hdfs.impl=$TACHYON_UNDERFS_HDFS_IMPL
  -Dtachyon.data.folder=$TACHYON_UNDERFS_ADDRESS/user/spark/tach_data
  -Dtachyon.workers.folder=$TACHYON_UNDERFS_ADDRESS/user/spark/tach_worker
  -Dtachyon.worker.memory.size=$TACHYON_WORKER_MEMORY_SIZE
  -Dtachyon.worker.data.folder=$TACHYON_RAM_FOLDER/tachyonworker/
  -Dtachyon.master.worker.timeout.ms=60000
  -Dtachyon.master.hostname=$TACHYON_MASTER_ADDRESS
  -Dtachyon.master.journal.folder=$TACHYON_UNDERFS_ADDRESS/user/spark/tach_journal/  
  -Dorg.apache.jasper.compiler.disablejsr199=true
  -Dtachyon.usezookeeper=true
  -Dtachyon.zookeeper.address=hdp02:2181,hdp03:2181,hdp04:2181
  -Dtachyon.zookeeper.election.path="/tach_elect"
  -Dtachyon.zookeeper.leader.path="/tach_leader"
  -Djava.net.preferIPv4Stack=true
"
# Master specific parameters. Default to TACHYON_JAVA_OPTS.
export TACHYON_MASTER_JAVA_OPTS="$TACHYON_JAVA_OPTS"
# Worker specific parameters that will be shared to all workers. Default to TACHYON_JAVA_OPTS.
export TACHYON_WORKER_JAVA_OPTS="$TACHYON_JAVA_OPTS"
---------------------------------------------------------------------------------------------------------------
vim slaves
hdp01
hdp02
hdp03
hdp04

第一次运行需要format

$ tachyon format
......
#
Connection to hdp01... Formatting Tachyon Worker @ hdp01
Removing local data under folder: /var/lib/spark/tachyon-0.5.0/ramdisk/tachyonworker/
Connection to hdp01 closed.
Connection to hdp02... Formatting Tachyon Worker @ hdp02
Removing local data under folder: /var/lib/spark/tachyon-0.5.0/ramdisk/tachyonworker/
Connection to hdp02 closed.
Connection to hdp03... Formatting Tachyon Worker @ hdp03
Removing local data under folder: /var/lib/spark/tachyon-0.5.0/ramdisk/tachyonworker/
Connection to hdp03 closed.
Connection to hdp04... Formatting Tachyon Worker @ hdp04
Removing local data under folder: /var/lib/spark/tachyon-0.5.0/ramdisk/tachyonworker/
Connection to hdp04 closed.
......

启动tachyon,为了让ramFS能mount，需要使用root来启动。

tachyon-start.sh all Mount
Starting master @ hdp04
...
Connection to hdp04... Formatting RamFS: /var/lib/spark/tachyon-0.5.0/ramdisk (10gb)
Starting worker @ hdp04
...

打开web UI http://hdp04:19999/home, 可以看到可用内存为我们配置的TACHYON_WORKER_MEMORY_SIZE大小。
tachyon-ui

在Spark中使用Tachyon

首先在$SPARK_HOME/conf中新建core-site.xml文件

    fs.tachyon.impl
    tachyon.hadoop.TFS

然后$SPARK_HOME/conf/spark-env.sh中设置
export SPARK_CLASSPATH=/var/lib/spark/tachyon-0.5.0/client/target/tachyon-client-0.5.0-jar-with-dependencies.jar:$SPARK_CLASSPATH
下面在Spark程序中需要指定：
spark.tachyonStore.url
spark.tachyonStore.BaseDir
例如：

vim $SPARK_HOME/conf/spark-defaults.conf
spark.tachyonStore.url tachyon://hdp4:19998
spark.tachyonStore.baseURL /data/tach_tmp
#打开Spark-Shell,通过tachyon来加载underfs（HDFS）中的数据/user/spark/1.txt
scala> val mydata = sc.textfile("tachyon-fs://hdp04:19998/user/spark/1.txt")
scala> data.count
...
res0: Long = 52
# 以in-memory保存到tachyon，可供其他应用使用
scala> mydata.saveAsTextFile("tachyon-fs://hdp04:19998/my/1.txt")
这个文件就以in-memory file保存到tft中，可以在web UI中查看，并通过命令进行删除
tachyon tft rm /my/1.txt
#使用tachyon持久化RDD，由于之前已经设置了spark.tachyonStore.url和spark.tachyonStore.baseDir，可以直接使用spark sc的persist(StorageLevel.OFF_HEAP)来持久化，当该Spark SC结束时，RDD会被自动清理。
scala> mydata.persist(StorageLevel.OFF_HEAP)

underfs和ramfs之间的数据同步

当TACHYON启动时，Tachyon不会知道已经存在的数据文件。需要使用下面的同步数据命令
$ ./bin/tachyon loadufs [TACHYON_PATH] [UNDERLYING_FILESYSTEM_PATH] [Optional EXCLUDE_PATHS]
例如:
$ tachyon loadufs tachyon://hdp04:19998 hdfs://hdp01:8020/user/spark/tach_data
第三个可选参数指的是UNDERLYING_FILESYSTEM_PATH下的这个PATH列表会除外，不加载到tfs中。
例如”tachyon;spark”表示hdfs://hdp01:8020/user/spark/tach_data下的tachyon和spark目录不会被加载。

tachyon的命令行参数

参照官方文档：http://tachyon-project.org/Command-Line-Interface.html
tachyon的命令行操作和hdfs类似，除了文件系统操作ls, lsr, mkdir, rm, mv, copyFromLocal, copyToLocal还一些工具命令cat, count(Display the number of folders and files matching the specified prefix in “path”.), tail(Print the last 1KB of the specified file to the console.), touch, fileinfo(Print the information of the blocks of a specified file.)之外，还有特有的pin和unpin命令：
command: pin
usage:pin “path”
Description:Pins the given file, such that Tachyon will never evict it from memory. If called on a folder, it recursively pins all contained files and any new files created within this folder.
command: unpin
usage: unpin “path”
Description:Unpins the given file to allow Tachyon to start evicting it again. If called on a folder, it recursively unpins all contained files and any new files created within this folder.

^o^