azkaban源码：azkaban-common:Project
由于种种原因，要基于azkaban二次开发一个适合自身的调度框架。所以进行源码的学习。方便后续开发。
源码构建
- build
- gradlew clean
- Gradlew build
- gradlew installDist
- create db
- CREATE DATABASE azkaban;
- run sql script
- download jdbc jar
Project
该模块主要包含了project的一系列操作。
假设我们现在有一个项目的zip文件。如何解决如下问题：
- 如何创建一个project？
- 一个project里面包含flow。这个flow是如何构建的？
- 一个flow里面应该有多个job或者说node。又是如何构建的？
- project flow 这些概念持久化存储在哪？
- 如何查询project flow 信息？
- 一个project中的flow如何调度？
当然这里面也包含了user，permission等概念。
首先 我们来看一下项目创建的流程。这里面只是创建project。还没有上传zip。
- User 提交数据
- ProjectManagerServlet 处理请求。
- 调用ProjectManager创建project。
- 调用ProjectLorder创建project。
- 调用ProjectLorder实现类JDBCProjectImpl创建project
- 调用DataBaseOperator提供的方法插入到数据库。
接下来 我们来看一下项目上传的流程。==解析flow等信息都是在上传的过程中完成的。==
- user提交zip
- ProjectManagerServlet处理请求
- 调用ProjectManager 上传project
- 调用AzkabanProjectLoader上传project
- 解压zip
- 调用FlowLoaderFactory创建FlowLoader，创建Flow，并将Flow设置为Project里面的属性。
- 调用StorageManager，内部调用projectloader上传信息到db。
- 调用projectLoader，执行各种方法，持久化相关信息到DB。
最后看一下其他查询请求的处理。
handleGet
- handleAJAXAction
- ajaxFetchProjectLogEvents: projectManager.getProjectEventLogs(project, num, skip); 获取项目日志信息
- ajaxFetchFlow
- ajaxFetchFlowDetails
- ajaxFetchFlowGraph
- ajaxFetchFlowNodeData
- ajaxFetchProjectFlows
- ajaxChangeDescription
- ajaxGetPermissions
- ajaxGetGroupPermissions
- ajaxGetProxyUsers
- ajaxChangePermissions
- ajaxAddPermission
- ajaxAddProxyUser
- ajaxRemoveProxyUser
- ajaxFetchFlowExecutions
- ajaxFetchLastSuccessfulFlowExecution
- ajaxFetchJobInfo
- ajaxSetJobOverrideProperty
- handleProjectLogsPage
- handlePermissionPage
- handlePropertyPage
- ...
基本上类似上述过程。再次不做重复。
通过上述的表述，我们可以发现。主要的project处理的的逻辑都集中在ProjectManager上。下面我们将详细的学习分析ProjectManager。
ProjectManager
project层次上的管理。单例模式。
==主要包含各种姿势的project的查询设置创建等。属于对外暴漏提供服务的类==。里面内部实现是由AzkabanProjectLoader，ProjectLoader等类提供服务实现的。
getProjectNames()
getUserProjects()
getGroupProjects()
getUserProjectsByRegex()
getProjects()
getProjectsByRegex()
isActiveProject()
getProject()
createProject()
purgeProject()
removeProject()
updateProjectDescription()
...
整个调用的层次结构大概如下：
- projectManager
- AzkabanProjectLoader
- ProjectLoader
- StorageManager
- ProjectLoader
- Storage
- FlowLoaderFactory
- FlowLoader
- ProjectLoader
- JdbcProjectImpl
- ...
- AzkabanProjectLoader
AzkabanProjectLoader
Handles the downloading and uploading of projects.
主要处理project的上传和下载。
核心方法==uploadProject==过程。
- 解压缩 file = unzipProject(archive, fileType);
- 验证有效性 reports = validateProject(project, archive, file, prop);
- 创建flowloader loader = this.flowLoaderFactory.createFlowLoader(file);
- ==工厂模式==：根据AZKABAN_FLOW_VERSION创建不同的flowloader，即DirectoryYamlFlowLoader或者DirectoryFlowLoader
- 上传project到db persistProject(project, loader, archive, file, uploader);
- 清空目录 FlowLoaderUtils.cleanUpDir(file);
- 返回reports
ProjectLoader
定义了各种接口。其应该有各种实现，默认使用jdbcprojectimpl实现。
JdbcProjectImpl
ProjectLoader的一种实现，This class implements ProjectLoader using new azkaban-db code to allow DB failover. 单例模式。
其主要功能是提供存储project各种信息的持久化方法。以及这新信息获取的方法。
具体的方法：
- fetchAllActiveProjects：获取所有active的项目
- fetchProjectById：通过projectid 获取project
- fetchProjectByName：通过project name 获取project
- createNewProject：根据名字 描述等创建新项目.
- uploadProjectFile: 上传项目文件，这个函数挺有意思的。里面是读这个文件，然后将它分成每10M一个chuck，然后插入到数据库中，并更新相应的表。
- addProjectVersion: 添加projectversion
- fetchProjectMetaData：获取上传的项目元数据。返回类型是ProjectFileHandler，具体数据为里面的属性。
- getUploadedFile：获取上传的file，返回类型是ProjectFileHandler，具体数据为里面的属性。其中查处所有的chuck，然后写入到文件中，再计算生成文件的md5与元数据中的md5是否一致，一直返回文件。不一致抛出异常。
- changeProjectVersion：修改项目版本。
- updateProjectSettings：更改项目配置。
- removeProject：为假删除。SET active=false
- postEvent
- getProjectEvents
- updateDescription
- getLatestProjectVersion
- uploadFlows： 该方法主要是上传一个project的flow。首先讲一个flow对象 转换成一个Map<String, Object>，再然后转换成json，再然后转换成byte[] 存入到数据库中。
- fetchAllProjectFlows：查询一个project的所有flow。
- uploadProjectProperty：上传project properties。
- fetchProjectProperty
- cleanOlderProjectVersion
- uploadFlowFile 上传flow file 类似于uploadFlows
- getUploadedFlowFile 获取flow file 类似于fetchAllProjectFlows
- getLatestFlowVersion
- isFlowFileUploaded
StorageManager
StorageManager manages and coordinates all interactions with the Storage layer. This also includes bookkeeping like updating DB with the new versionm, etc
azkaban提供了多种存储层的实现方式，包括database，hdfs，localfile等。也可以自定义自己的存储层。默认使用的是database进行存储。
其中这里面最重要的对外的类应该是==StorageManager==。主要提供了uploadproject和getprojectfile等方法。
FlowLoaderFactory
Factory class to generate flow loaders.
flowLoader
Interface to load project flows.
包含以下重要接口：
- ValidationReport loadProjectFlow(final Project project, final File projectDir);
- Map<String, Flow> getFlowMap();
更直白一点，就是提供如何从一个project中构建flow的功能。
根据不同版本，flowloader提供了两种实现方式。
- DirectoryYamlFlowLoader：flow信息存储在yaml文件中。
- DirectoryFlowLoader：flow信息以directory的方式存在。
NodeBeanLoader
其实该部分 应该是如何构建flow，包括node，什么的。最终对外输出的是==FlowLoaderUtils==。里面包含了yaml方式的构建和一些公用的方法。
- AzkabanNode
- A unit of execution that could be either a job or a flow.
- AzkabanFlow
- Flow level definition of the DAG.
- Contains a list of AzkabanNodes and related flow properties.
- AzkabanJob
- The smallest individual unit of execution in Azkaban.
- Contains information about job type and related properties.
数据结构
- node的内部属性：
- level: 距离起始节点的距离（多条路径的话取最大值），起始节点level为0
- jobSource
- propsSource
- Edge的内部属性：
- id：getSourceId() + ">>" + getTargetId()
- sourceId
- source
- targetId
- target
- error
- flow的内部属性：
- inEdges：key为jobName，value为节点的入边集合
- outEdges：key为jobName，value为节点的出边集合
- startNodes：起始Node集合，没有入边的谓之起始节点
- endNodes：结束Node结合，没有出边的谓之结束节点
- 每个flow里能配置报警等信息
AZ.Storage
Storage:
/**
* The Azkaban Storage interface would facilitate getting and putting objects into a storage
* mechanism of choice. By default, this is set to the MySQL database. However, users can have the
* ability to choose between multiple storage types in future.
*
* This is different from storing Azkaban state in MySQL which would typically be maintained in a
* different database.
*
* Note: This is a synchronous interface.
*/
Az.DB
该模块主要是azkaban db相关代码。包含以下功能。
- 建立jdbc连接，获取datasource
- 执行sql操作接口。
如需扩展其他链接，需要继承==AzkabanDataSource==。实现相应功能。另外，对外使用 主要使用AzkabanDataSource 和DatabaseOperator两个类。
