大数据工做流开源系统之DRAKE
1工做流开源系统概要
如下是各种数据分析的工做流pipeline管道模型的框架和库包，包含特定科学等业务领域行业或者通用领域，结合大数据big data分析的各类开放源码项目。包括下一代测序技术(NGS)打开了数据分析的空间，生物学成为数据密集领域，愈来愈多的生物数据须要经过复杂的计算工具（集群、云端和网格计算）进行NGS处理和分析。html
| 工做流开源系统 | 网址 | 现类 |
|---|---|---|
| Arvados | http://arvados.org | 产品级的数据科学平台，能处理大数据集，由两个主要系统组成以及涵盖不少相关服务与组件，包括API SDK和虚拟化工具。 |
| Taverna | http://www.taverna.org.uk/ | 一个与具体领域无关的独立的工做流管理系统，包括一系列工具用于设计和科学工做流程的实现。 |
| Galaxy | http://galaxyproject.org/ | 基于web用于数据密集的生物医学研究 |
| SHIWA | http://www.erflow.eu/ | 跨国的欧洲工做流用户社区推出的用于科学研究的工做流。 |
| Oozie | https://oozie.apache.org/ | 是一个用于管理Apache Hadoop job的工做流调度系统。 |
| DNANexus | https://wiki.dnanexus.com/API-Specification-v1.0.0/IO-and-Run | 不只能在云端存储和分析文件，也有助于协助 注释和虚拟化你的数据。 |
| BioDT | http://www.biodatomics.com/ | 下一代生物信息学工具 |
| Agave | http://agaveapi.co/live-docs/ | 一个Science-as-a-Service API平台 |
| DiscoveryEnvironment | http://www.iplantcollaborative.org/ci/discovery-environment | iPlant 的信息基础平台 |
| Wings | http://www.wings-workflows.org/ | 是一个语义工做流系统，辅助科学家进行计算体验设计。 |
| Knime | https://www.knime.org/ | 强大的大数据分析和挖掘能力。 |
| Drake | https://github.com/Factual/drake | 易于使用可扩展 基于文本的数据工做流工具，组织围绕数据的命令行执行，相似GNU的 Make |
| Snakemake | https://bitbucket.org/johanneskoester/snakemake | 是一个工做流管理系统，下降建立工做流的难度复杂性，提供快速舒服的执行环境。 |
| BPipe | http://bpipe.org | 用于生物学领域的任务运行，用以替代各类脚本任务。 |
| Ruffus | https://code.google.com/p/ruffus/ | Python可计算的Pipeline，用于生物信息学在并行流程中管理依赖。 |
| NextFlow | http://nextflow.io | 借鉴Unix管道模型，提供一个流畅的DSL，容许你处理复杂的流交互。 |
| Luigi | http://github.com/spotify/luigi | Luigi是一个Python (2.7, 3.3, 3.4, 3.5)库包，帮助你创建复杂的批工做的管道模型pipeline，它处理依赖问题，工做流管理，虚拟化，处理失败，集成命令行等不少功能。 |
| SciLuigi. | http://github.com/samuell/sciluigi | |
| GATK Queue | https://www.broadinstitute.org/gatk/guide/topic?name=queue | 创建一个分析管道模型有效率运行GATK和其余工具 |
| Yabi | https://ccg.murdoch.edu.au/yabi | |
| seqware | https://seqware.github.io/ | https://seqware.github.io/docs/6-pipeline/ |
| Ketrew | https://github.com/hammerlab/ketrew | 一个提供EDSL API的OCaml库 |
| Pegasus | http://pegasus.isi.edu/ | 工做流管理系统，自动恢复和调式科学计算。 |
| Airflow | https://github.com/airbnb/airflow（重复） | |
| Cosmos/ Cosmos2: | https://github.com/LPM-HMS/COSMOS2 | 主流的并行工做流Python库包，生物科学领域下一代序列流管道模型，提供命令行工具自动利用计算集群能力，提供Web仪表板监视，调试，分析你的job。 |
| Pinball | https://github.com/pinterest/pinball | Pinball是个可伸缩扩展的工做流管理器，基于组件，状态以可读格式保存，可靠，管理友好。 |
| bcbio | https://bcbio-nextgen.readthedocs.org/en/latest/ | 提供高吞吐量数据序列流分析，只要编写高层配置指定输入和分析参数，输入会驱动并行管道模型处理被分发的执行任务，幂等处理可从新启动，透明的事务步骤。 |
| Chronos | https://github.com/mesos/chronos | Chronos是一个linux cron的替代，它是分布式的失败容错的调度器，运行在 Apache Mesos之上，支持定制优化Mesos执行器做为默认的命令执行器。 |
| Azkaban | https://azkaban.github.io/ | 是一个批工做流任务调度器，由LinkedIn建立，用于运行Hadoop job，解决了job的依赖顺序，提供易于使用的Web用户界面维护和跟踪你的处理流程。 |
| Apache NiFi | https://nifi.apache.org/docs/nifi-docs/html/overview.html | 系统之间数据流程的自动化，当数据流使用在各类场合，使用它自动化和管理系统之间信息流动。 |
| flowr (R-based) | http://docs.flowr.space/ | Flowr让你设计和实现复杂的管道模型，部署在你的计算集群，知足生物信息学须要。 |
| Mistral | https://github.com/arteria-project | 从主流并行序列流中处理序列数据，提供组件实现自动分析和数据管理任务做为下一代序列流中心，平衡微服务架构，使用StackStorm建立一个事件驱动的自动化系统。灵活可伸缩。 |
| nipype | http://nipy.org/nipype/ | |
| End of Day | https://github.com/joestubbs/endofday | docker 容器的可执行工做流，可使用yaml文件定义。 |
| BioDSL | https://github.com/maasha/BioDSL | 用于生物学的领域特定语言。 |
| BigDataScript | http://pcingola.github.io/BigDataScript/ | 在笔记本上开发的数据分析管道模型能够运行在大数据几千节点集群上。 |
| Omics Pipe: | http://sulab.scripps.edu/omicspipe/ | 开源模块计算平台，自动化multi-omics数据分析管道模型的最佳实践。 |
| Ensembl Hive | https://github.com/Ensembl/ensembl-hive | eHive是一个在分布式计算资源运行计算管道模型的系统。 |
| QuickNGS | http://bifacility.uni-koeln.de/quickngs/web | 下一代测序数据的高吞吐量的数据分析。 |
| GenePattern | http://www.broadinstitute.org/cancer/software/genepattern/ | 提供数百个分析工具，分析 gene expression (RNA-seq 和 microarray), sequence variation and copy number, proteomic, flow cytometry, 和网络分析 |
| Chipster | http://chipster.csc.fi/ | 新一代测序(NGS)，提供超过350分析工具 |
| The Genome Modeling System | https://github.com/genome/gms | |
| Cuneiform, | https://github.com/joergen7/cuneiform | 函数式工做流语言 |
| Anvaya | http://www.ncbi.nlm.nih.gov/pubmed/22809419 | 自动化基因分析的工做流环境 |
| Makeflow | http://ccl.cse.nd.edu/software/makeflow/ | 在集群 云端和网格之上执行大型复杂工做流引擎 |
| Airavata | http://airavata.apache.org/ | 基于集群，云端和网格之上组合，管理，执行和监视大规模可伸缩应用和工做流引擎 |
| Pyflow | https://github.com/Illumina/pyflow | 一个轻量的并行任务引擎 |
| Clusterflow | https://github.com/ewels/clusterflow | 一个管道模型工具，在集群环境自动化和标准化生物分析。 |
| Unipro UGENE | http://ugene.net/ | https://dx.doi.org/10.7717/peerj.644 |
| CloudSlang | http://www.cloudslang.io/ | 管理协调Docker和CoreOS应用，快速化自动DevOps |
| Stacks | http://catchenlab.life.illinois.edu/stacks/ | 用于构建Ioci的软件管道模型 |
| Leaf | http://www.francesconapolitano.it/leaf/index.html | 用于数据流Bioinformatic Protocol的设计和管理的Python工具 |
| omictools | http://omictools.com/ | 提供11573 种分析工具 |
| Job 描述语言 | https://edms.cern.ch/ui/file/590869/1/WMS-JDL.pdf | |
| YAWL | http://www.yawlfoundation.org/ | 一个BPM/工做流系统，基于简明和强大的模型语言，处理复杂数据转换，彻底集成各类Web服务。 |
| Triquetrum | https://github.com/eclipse/triquetrum/ | 科学工做流的管理和执行。 |
| Kronos | https://github.com/jtaghiyar/kronos | 癌症和肿瘤信息分析 |
| qsubsec | https://github.com/alastair-droop/qsubsec | 基于SGE grid system的模板语言，产生用于提交任务的脚本语言。 |
| YesWorkflow | http://yesworkflow.org | 将工做流模型带到脚本语言中 |
| GWF - Grid WorkFlow | https://github.com/mailund/gwf | 小的相似make的经过qsub提交工做流的工具。 |
| Fireworks | https://pythonhosted.org/FireWorks/ | 是一个定义 管理和执行工做流的框架，能使用 Python, JSON, 或 YAML定义复杂工做流，存储在MongoDB中，可以经过Web界面监视，工做流执行能自动跨无数个计算资源执行。 |
| NGLess | https://github.com/luispedro/ngless | 是下一代测序NGS 的领域特定语言。 |
2咱们选用了drake
Drake--易于使用可扩展 基于文本的数据工做流工具，组织围绕数据的命令行执行，相似GNU的 Make java
【项目地址】https://github.com/Factual/drake
【项目简介】清理脏数据，输入输出更可控一些。
【入门例子】https://github.com/Factual/drake/wiki/Tutorialpython
简单例子：
ruichao-factual编辑这个页面on 25 Nov 2013 ·linux
本教程是一项正在进行中的工做。 若是有一个特定的话题你想覆盖,请经过 Google Group for Drake.让咱们知道。git
概述
你工程所写的Drake workflow文件将指定运行哪些步骤。 通常来讲,每个step都依赖于一个或多个输入源,并将建立一个或多个输出组件。github
Drake workflow文件经过step来组织。 除了指定输入和输出,一个step一般会包含明确的命令,可能还有额外的选项。web
这里有一个案例是一个简单的step:docker
; we only like lines with lowercase "i" in them out.csv <- in.csv [shell] grep i $INPUT > $OUTPUT
上面的步骤使用drake的“shell”协议,这意味着命令为shell命令。 (也能够有其余的协议,必须显式地指定。 但在本教程中,咱们将主要关注如何使用shell协议。)shell
让咱们逐步的分解上述步骤的特定参数:express
- out.csv:生产输出文件
- in.csv: 要使用的输入文件
- [shell]: 括号括起来的可选项. 对于step的协议来讲是一个很是重要的选项. 在这里,咱们选择了“shell”协议,它容许咱们在这个步骤运行shell命令。
- the indented line:缩进线如下的第一行是一个step的命令。 在这种状况下,有一个命令,该命令执行行过滤。注意,命令shell命令,代表每次执行都将使用shell协议。
- $INPUT: 一个Drake shell step将在运行shell命令的step以前，自动加载shell环境变量和有用的信息. 例如,它加载的文件路径输入环境变量指定的输入的第一步。 所以,步骤的shell命令访问变量,好比 $INPUT.
- $OUTPUT: 相似于 $INPUT,一个 Drake shell step 将在运行shell以前，自动加载输出的文件路径环境变量到指定的输出.
基本的依赖关系管理
一个Drake workflow可能会有不少的step,他们可能会以各类方式相互依赖。 当咱们考虑了这个额外的状况事，上面的例子将被增长：
; produce an extraordinarily fancy report count.txt <- out.csv wc $INPUT > $OUTPUT
这一步取决于out.csv(也就是说,它使用out.csv做为它的输入文件),并产生count.txt。 由于依赖out.csv,Drake将默认状况下肯定out.csv是最新的。 这意味着若是有必要的话，Drake将运行所需的step(s)建立out.csv)。 (这种行为是基本依赖关系管理的宗旨,这很像咱们知道的也喜欢使用的工具--Make。)
Drake的命令行接口容许咱们首先执行哪一个特定的step,也有其余各类命令选项。 可是，默认状况下,Drake将尝试在您的workflow运行的全部steps。
有关Drake命令行选项的更多细节,包括选项,请参阅完整的用户手册。
可是咱们要提早。 让咱们经过如下例子来学习……
你的第一个工做流
Drake是创建工做流运行数据。 默认状况下,它在寻找工做流程文件在路径./Drakefile。 这就是为何若是你运行它在没有./Drakefile文件的地方时，Drake有时会报错找不到工做流程文件。
让咱们开始一个新的工做流程,在一个新目录:
$ mkdir /myworkflow $ cd /myworkflow
如今建立一个简单的工做流。 建立一个文件命名workflow.d和把这个(前面示例拷贝便可):
; we only like lines with lowercase "i" in them out.csv <- in.csv grep i $INPUT > $OUTPUT
这是一个很是简单的Drake workflow,只有一个步骤。 这个step运行一个shell命令,使用in.csv输入文件和输出写入output.csv。
咱们尚未一个输入文件,因此让咱们建立它。 建立一个文件命名in.csv和而后发那个放入一些测试数据,以下所示:
Artem,Boytsov,artem Aaron,Crow,aaron Alvin,Chyan,alvin Maverick,Lou,maverick Vinnie,Pepi,vinnie Will,Lao,will
酷,如今咱们有一个Drake workflow和一个简单的输入文件运行工做流。 让咱们运行它!
$ drake -w workflow.d
让咱们检查输出:
$ more out.csv Alvin,Chyan,alvin Maverick,Lou,maverick Vinnie,Pepi,vinnie Will,Lao,will
3你觉得这就完了？No，下面是干货
workflow_douyu_data.d
; load the plugin defination
%include $[PROFILE]
; script
plugin_version="1.0"
plugin_home=/var/weiboyi/azkaban/UnifiedDataProcessing/douyuCrawlParser
plugin_user_info_generator=$[plugin_home]/app/$[plugin_version]/bin/douyu_account_info_generator.sh
plugin_media_info_generator=$[plugin_home]/app/$[plugin_version]/bin/douyu_media_info_generator.sh
plugin_user_info_config_generator=$[plugin_home]/app/$[plugin_version]/config/douyu_account_generator_pig.params
plugin_media_info_config_generator=$[plugin_home]/app/$[plugin_version]/config/douyu_media_generator_pig.params
; hdfs crawled data path
hdfs_douyu_crawled_info_path=/tech/azkaban/data_processing/crawler2/douyu
; [account_input_data] hdfs crawled account data path
hdfs_account_info_path=$[hdfs_douyu_crawled_info_path]/account_info
hdfs_account_info_data_tag=$[hdfs_account_info_path]/data/douyu_account_info
; [media_input_data] hdfs crawled media data path
hdfs_media_info_path=$[hdfs_douyu_crawled_info_path]/media
hdfs_media_info_data_tag=$[hdfs_media_info_path]/data/douyu_media
account_info_latest_path=$($[plugin_utils_hdfs_interaction_with_local_file] --interaction_mode=print --data_hdfs_path=$[hdfs_account_info_data_tag] --item_count=1 --data_hdfs_path_list)
account_info_latest_date=$($[plugin_utils_hdfs_interaction_with_local_file] --interaction_mode=print --data_hdfs_path=$[hdfs_account_info_data_tag] --item_count=1 --valid_time)
account_info_schema_path=$($[plugin_utils_hdfs_interaction_with_local_file] --interaction_mode=print --data_hdfs_path=$[hdfs_account_info_data_tag] --item_count=1 --schema_hdfs_path)
media_info_latest_path=$($[plugin_utils_hdfs_interaction_with_local_file] --interaction_mode=print --data_hdfs_path=$[hdfs_media_info_data_tag] --item_count=1 --data_hdfs_path_list)
media_info_latest_date=$($[plugin_utils_hdfs_interaction_with_local_file] --interaction_mode=print --data_hdfs_path=$[hdfs_media_info_data_tag] --item_count=1 --valid_time)
media_info_schema_path=$($[plugin_utils_hdfs_interaction_with_local_file] --interaction_mode=print --data_hdfs_path=$[hdfs_media_info_data_tag] --item_count=1 --schema_hdfs_path)
; schema version
account_info_schema_version=1.0
media_info_schema_version=1.0
; hdfs output data path
hdfs_feature_data_path=/tech/azkaban/data_processing/feature_generator/douyu/data
hdfs_account_info_snapshot_path=$[hdfs_feature_data_path]/account_info_snapshot
hdfs_account_info_snapshot_schema_path=$[hdfs_account_info_snapshot_path]/schema
hdfs_account_info_snapshot_data_tag=$[hdfs_account_info_snapshot_path]/data/douyu_account_info_snapshot
hdfs_media_snapshot_path=$[hdfs_feature_data_path]/media_info_snapshot
hdfs_media_snapshot_schema_path=$[hdfs_media_snapshot_path]/schema
hdfs_media_snapshot_data_tag=$[hdfs_media_snapshot_path]/data/douyu_media_info_snapshot
account_info_snapshot_latest_path=$($[plugin_utils_hdfs_interaction_with_local_file] --interaction_mode=print --data_hdfs_path=$[hdfs_account_info_snapshot_data_tag] --item_count=1 --data_hdfs_path_list)
account_info_snapshot_latest_date=$($[plugin_utils_hdfs_interaction_with_local_file] --interaction_mode=print --data_hdfs_path=$[hdfs_account_info_snapshot_data_tag] --item_count=1 --valid_time)
account_info_snapshot_schema_path=$($[plugin_utils_hdfs_interaction_with_local_file] --interaction_mode=print --data_hdfs_path=$[hdfs_account_info_snapshot_data_tag] --item_count=1 --schema_hdfs_path)
media_snapshot_latest_path=$($[plugin_utils_hdfs_interaction_with_local_file] --interaction_mode=print --data_hdfs_path=$[hdfs_media_snapshot_data_tag] --item_count=1 --data_hdfs_path_list)
media_snapshot_latest_date=$($[plugin_utils_hdfs_interaction_with_local_file] --interaction_mode=print --data_hdfs_path=$[hdfs_media_snapshot_data_tag] --item_count=1 --valid_time)
media_snapshot_schema_path=$($[plugin_utils_hdfs_interaction_with_local_file] --interaction_mode=print --data_hdfs_path=$[hdfs_media_snapshot_data_tag] --item_count=1 --schema_hdfs_path)
; douyu account info snapshot
hdfs:$[hdfs_account_info_snapshot_data_tag]_v$[account_info_schema_version]_$[account_info_latest_date], hdfs:$[hdfs_account_info_snapshot_schema_path]/schema_douyu_account_info_snapshot_v$[account_info_schema_version] <- hdfs:$[account_info_latest_path], hdfs:$[account_info_schema_path]
hdfs_crawl_account_data=$($[plugin_utils_hdfs_interaction_with_local_file] --interaction_mode=print --data_hdfs_path=$[hdfs_account_info_data_tag] --start_time=$[account_info_snapshot_latest_date] --end_time=$[account_info_latest_date] --data_hdfs_path_list)
cmd="/bin/bash $[plugin_user_info_generator] --account_info_data_path ${hdfs_crawl_account_data} --account_info_schema_path $INPUT1 --last_account_info_snapshot_data_path $[account_info_snapshot_latest_path] --last_account_info_snapshot_schema_path $[account_info_snapshot_schema_path] --cur_account_info_snapshot_data_path $OUTPUT0 --account_info_snapshot_schema_path $OUTPUT1 --account_info_config_path $[plugin_user_info_config_generator]"
echo "$cmd"
eval "$cmd"
if [ $? -eq 0 ]; then
hadoop fs -touchz $[hdfs_account_info_snapshot_path]/data/available_$[account_info_latest_date]
else
echo "$cmd is failed,please check!" >> ${send_emails_file}
fi
; douyu media info snapshot
hdfs:$[hdfs_media_snapshot_data_tag]_v$[media_info_schema_version]_$[media_info_latest_date], hdfs:$[hdfs_media_snapshot_schema_path]/schema_douyu_media_info_snapshot_v$[media_info_schema_version] <- hdfs:$[media_info_latest_path], hdfs:$[media_info_schema_path]
hdfs_crawl_media_data=$($[plugin_utils_hdfs_interaction_with_local_file] --interaction_mode=print --data_hdfs_path=$[hdfs_media_info_data_tag] --start_time=$[media_snapshot_latest_date] --end_time=$[media_info_latest_date] --data_hdfs_path_list)
cmd="/bin/bash $[plugin_media_info_generator] --media_info_data_path ${hdfs_crawl_media_data} --media_info_schema_path $INPUT1 --last_media_info_snapshot_data_path $[media_snapshot_latest_path] --last_media_info_snapshot_schema_path $[media_snapshot_schema_path] --cur_media_info_snapshot_data_path $OUTPUT0 --media_info_snapshot_schema_path $OUTPUT1 --media_info_pig_config_path $[plugin_media_info_config_generator]"
echo "$cmd"
eval "$cmd"
if [ $? -eq 0 ]; then
hadoop fs -touchz $[hdfs_media_snapshot_path]/data/available_$[media_info_latest_date]
else
echo "$cmd is failed,please check!" >> ${send_emails_file}
fi
workflow.sh
#!/usr/bin/env bash
source '/var/weiboyi/azkaban/big_data/Common/workflow_environment_config.sh'
source '/var/weiboyi/azkaban/big_data/Common/workflow_plugin_definition.sh'
export PROFILE='/var/weiboyi/azkaban/big_data/Common/workflow_drake_profile'
export workflow_home=/var/weiboyi/azkaban/UnifiedDataProcessing/WorkFlow/Douyu/CrawledDataParser/trunk
data_folder='/data0/weiboyi/azkaban/UnifiedDataProcessing/WorkFlow/Douyu/CrawledDataParser/data'
workflow_name='douyuDataGenerator'
workflow_owner='yuanyihan@weiboyi.com'
today=`date +"%Y%m%d"`
drake_base_folder=${data_folder}/${today}
mkdir ${drake_base_folder}
statistic_result=${drake_base_folder}/statistic_result
if [ -f ${statistic_result} ]; then
rm ${statistic_result}
fi
send_emails_file=${drake_base_folder}/send_emails_result
if [ -f ${send_emails_result} ]; then
rm ${send_emails_file}
fi
# clear the expired data in data folder
find ${data_folder} -type d -mtime +30 -exec rm -rf {} \;
echo '#############################################################'
valid_date=$1
douyu_account_info_path=/tech/azkaban/data_processing/crawler2/douyu/account_info
douyu_account_info_tag=${douyu_account_info_path}/data/douyu_account_info
douyu_account_latest_date=$(${plugin_utils_hdfs_interaction_with_local_file} --interaction_mode=print --data_hdfs_path=${douyu_account_info_tag} --item_count=1 --valid_time)
echo "douyu_account_latest_date: ${douyu_account_latest_date}"
douyu_media_info_path=/tech/azkaban/data_processing/crawler2/douyu/media
douyu_media_info_tag=${douyu_media_info_path}/data/douyu_media
douyu_media_latest_date=$(${plugin_utils_hdfs_interaction_with_local_file} --interaction_mode=print --data_hdfs_path=${douyu_media_info_tag} --item_count=1 --valid_time)
echo "douyu_media_latest_date: ${douyu_media_latest_date}"
flag="false"
unset douyu_account_feature_valid_date
if [ ! ${valid_date} ];then
douyu_account_feature_valid_date=${douyu_media_latest_date}
flag="true"
elif [ "${valid_date}" -gt "${douyu_account_latest_date}" ] && [ "${valid_date}" -gt "${douyu_media_latest_date}" ]; then
echo "valid_date ${valid_date} is greater than douyu_account_latest_date ${douyu_account_latest_date}, exit" >> ${send_emails_file}
else
douyu_account_feature_valid_date=${valid_date}
flag="true"
fi
echo "douyu_account_feature_valid_date: ${douyu_account_feature_valid_date}"
echo "##########################################################"
# call the drake workflow
if [ ${flag} == "true" ]; then
cmd="${drake_cmd_env} --auto --workflow=${workflow_home}/workflow_douyu_data.d --base=${drake_base_folder} --logfile=${drake_base_folder}/drake_${today}.log --tmpdir=${drake_base_folder}/drake_tmp -v statistic_result=${statistic_result},send_emails_file=${send_emails_file}"
echo "$cmd"
eval "$cmd"
fi
echo '############################### upload hbase start ##############################'
#upload DB
plugin_version="2.0"
plugin_home=/var/weiboyi/azkaban/UnifiedDataProcessing/AccountFeatureUpload
plugin_upload_hbase_generator=${plugin_home}/app/${plugin_version}/bin/account_feature_upload_hbase_java_generator.sh
plugin_upload_douyu_data_path=/tech/azkaban/data_processing/feature_generator/douyu/data/media_info_snapshot
hdfs_upload_douyu_schema_path=${plugin_upload_douyu_data_path}/schema
hdfs_upload_douyu_data_tag=${plugin_upload_douyu_data_path}/data/douyu_media_info_snapshot
plugin_upload_douyu_given_data_path=$(${plugin_utils_hdfs_interaction_with_local_file} --interaction_mode=print --data_hdfs_path=${hdfs_upload_douyu_data_tag} --start_time=${valid_date} --end_time=${valid_date} --data_hdfs_path_list)
echo "plugin_upload_douyu_given_data_path: ${plugin_upload_douyu_given_data_path}"
hdfs_upload_douyu_schema_path=$(${plugin_utils_hdfs_interaction_with_local_file} --interaction_mode=print --data_hdfs_path=${hdfs_upload_douyu_data_tag} --item_count=1 --schema_hdfs_path)
echo "hdfs_upload_douyu_schema_path: ${hdfs_upload_douyu_schema_path}"
data_folder='${plugin_home}/data'
workflow_home='/var/weiboyi/azkaban/UnifiedDataProcessing/WorkFlow/Douyu/CrawledDataParser'
config_folder='${workflow_home}/config'
if [ ${plugin_upload_douyu_given_data_path} ];then
cmd="/bin/bash ${plugin_upload_hbase_generator} --input_data_path ${plugin_upload_douyu_given_data_path} --input_schema_path ${hdfs_upload_douyu_schema_path} --config_file ${config_folder}/account_douyu_upload_hbase.yaml --tableName pinggu_dw:media_douyu --isCreateHbaseTable false"
echo "$cmd"
eval "$cmd"
if [ $? -ne 0 ];then
echo "${valid_date}_${today}:exe douyu media snapshot upload hbase error,please check!">>${send_emails_file}
fi
fi
echo '############################### upload hbase end ##############################'
echo "##########################################################"
if [ -f ${send_emails_file} ];then
cmd="mail -s 斗鱼爬虫数据处理：${valid_date}_${today} ${workflow_owner} < ${send_emails_file}"
echo "$cmd"
eval "$cmd"
fi
echo "##########################################################"
相关文章
- 1. 工做流开源系统--DRAKE
- 2. 开源OA系统启动:基础数据,工做流设计
- 3. 工做流开源系统--OOZIE
- 4. 数据库系统的工做流程
- 5. 大数据之Spark -- 工做流程图
- 6. 主流开源大数据工具
- 7. 开源大数据处理系统/工具大全
- 8. 大数据开源工具
- 9. NiFi-大数据工做流平台，资源大全
- 10. 大数据分析系统Hadoop的13个开源工具
- 更多相关文章...
- • Git 工作流程 - Git 教程
- • 操作系统（OS）平台 统计 - 浏览器信息
- • PHP开发工具
- • 互联网组织的未来：剖析GitHub员工的任性之源
相关标签/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
最新文章
欢迎关注本站公众号,获取更多信息
相关文章
- 1. 工做流开源系统--DRAKE
- 2. 开源OA系统启动:基础数据,工做流设计
- 3. 工做流开源系统--OOZIE
- 4. 数据库系统的工做流程
- 5. 大数据之Spark -- 工做流程图
- 6. 主流开源大数据工具
- 7. 开源大数据处理系统/工具大全
- 8. 大数据开源工具
- 9. NiFi-大数据工做流平台，资源大全
- 10. 大数据分析系统Hadoop的13个开源工具