Azkaban作为LinkedIn开源的任务流式管理工具，在工作中很大程度上被用到。但是，由于非国人开发，对中文的支持性很不好。大多数情况下，会出现几种乱码现象: - 执行内置脚本生成log乱码 - 直接command执行中文乱码 - 中文包名乱码等，其中对日常使用影响最大的就是日志乱码问题。不管是调度Hive、DataX还是Java程序，只要日志抛出来中文，中文都是乱码显示，将日志文件拷贝出来查看文件格式为GB2312，但是Linux系统编码和Azkaban日志编码明明设置的都是UTF-8，很疑惑，摸索许久，决定从源码入手开始层层解惑。
文中大部分内容从源码一步步进入解析，有经验的朋友可以跳至文末见具体解决方法。
根据页面获取日志的接口可以知道方法在 azkaban-web-server项目下package azkaban.webapp.servlet 下的方法handleAJAXAction，如下图 请求参数是fetchExecJobLogs
对应的处理方法为 ajaxFetchJobLogs(req, resp, ret, session.getUser(), exFlow)和ajaxFetchExecFlowLogs(req, resp, ret, session.getUser(), exFlow)
进入该方法后可以发现返回的data为经过 StringEscapeUtils.escapeHtml格式化过的，这就是引发乱码的原因之一。
改用commons-lang3下的方法可以解决这个问题，增加如下依赖后更新gradle项目，将此处StringEscapeUtils.escapeHtml(data.getData())更改为 org.apache.commons.lang3.StringEscapeUtils.escapeHtml3(data.getData())
// https://mvnrepository.com/artifact/org.apache.commons/commons-lang3
compile group: 'org.apache.commons', name: 'commons-lang3', version: '3.4'
修改后为
仔细研读代码可以发现其实这样还不够，读取日志的时候会先判断该任务是否正在执行，如果在执行的话就直接在服务器上暂存日志路径读取，否则是话就在mysql表读取，具体代码如下
一.本地文件读取的话是走http请求的，对应的为azkaban-exec-server下的包package azkaban.execapp下的ExecutorServlet
进入方法handleFetchLogEvent,一个是FlowLogs 一个是JobLogs ,
进入这两个方法后发现最终都是调用的FileIOUtils.readUtf8File()这个方法
读取日志的核心方法就是这个了
BufferedInputStream读取的是字节byte,因为一个汉字占两个字节,而当中英文混合的时候,有的字符占一个字节,有的字符占两个字节,所以如果直接读字节,而数据比较长,没有一次读完的时候,很可能刚好读到一个汉字的前一个字节,这样,这个中文就成了乱码,后面的数据因为没有字节对齐,也都成了乱码.所以我们需要用BufferedReader来读取,
它读到的是字符,所以不会读到半个字符的情况,不会出现乱码。
在这里修改方法后为一个新方法如下(编码方式GBK根据生成的日志编码格式设定)
public static LogData readUtf8FileEncode(final File file, final int fileOffset, final int length)
throws IOException {
final char[] chars = new char[length];
final FileInputStream fileStream = new FileInputStream(file); final long skipped = fileStream.skip(fileOffset);
if (skipped < fileOffset) {
fileStream.close();
return new LogData(fileOffset, 0, "");
} BufferedReader reader = null;
int read = 0;
try {
reader = new BufferedReader(new InputStreamReader(fileStream, "GBK"));
read = reader.read(chars);
} finally {
IOUtils.closeQuietly(reader);
} if (read <= 0) {
return new LogData(fileOffset, 0, "");
}
final byte[] buffer = new String(chars).getBytes();
final Pair<Integer, Integer> utf8Range = getUtf8Range(buffer, 0, read);
final String outputString = new String(buffer, utf8Range.getFirst(), utf8Range.getSecond(),
StandardCharsets.UTF_8); return new LogData(fileOffset + utf8Range.getFirst(), utf8Range.getSecond(), outputString);
}
在原来引用的地方替换为新的即可，如下图。
二.对于运行完成的，日志会被写入数据库，而且是经过压缩的，方法为 package azkaban.executor下的ExecutionLogsDao 搜索找到fetchLogs，在方法体中实例化了一个FetchLogsHandler
在FetchLogsHandler中核心方法为handle，在这个方法中可以发现这块会有解压缩的过程，开始怀疑是不是这块出现问题，测试后否定了这一猜想。
这里有解压，那在入库之前肯定会有压缩过程，再找到入库的方法（如何触发将本地文件写入到数据库在这里不讨论）uploadLogFile（ExecutionLogsDao方法中）
进入方法体后可以发现和之前类似，也是采用BufferedInputStream读取文件
修改后的方法为如下(编码方式GBK根据生成的日志编码格式设定)
private void uploadLogFileEncode(final DatabaseTransOperator transOperator, final int execId, final String name,
final int attempt, final File[] files, final EncodingType encType) throws SQLException {
// 50K buffer... if logs are greater than this, we chunk.
// However, we better prevent large log files from being uploaded somehow
final char[] buffer = new char[50 * 1024];
int pos = 0;
int length = buffer.length;
int startByte = 0;
try {
for (int i = 0; i < files.length; ++i) {
final File file = files[i];
// BufferedInputStream读取的是字节byte,因为一个汉字占两个字节,而当中英文混合的时候,有的字符占一个字节,
// 有的字符占两个字节,所以如果直接读字节,而数据比较长,没有一次读完的时候,很可能刚好读到一个汉字的前一个字节,
// 这样,这个中文就成了乱码,后面的数据因为没有字节对齐,也都成了乱码.所以我们需要用BufferedReader来读取,
// 它读到的是字符,所以不会读到半个字符的情况,不会出现乱码
BufferedReader reader = new BufferedReader(new InputStreamReader(new FileInputStream(file), "GBK"));
try {
int size = reader.read(buffer, pos, length);
while (size >= 0) {
if (pos + size == buffer.length) {
// Flush here.
uploadLogPart(transOperator, execId, name, attempt, startByte, startByte + buffer.length,
encType, buffer, buffer.length); pos = 0;
length = buffer.length;
startByte += buffer.length;
} else {
// Usually end of file.
pos += size;
length = buffer.length - pos;
}
size = reader.read(buffer, pos, length);
}
} finally {
IOUtils.closeQuietly(reader);
}
} // Final commit of buffer.
if (pos > 0) {
uploadLogPart(transOperator, execId, name, attempt, startByte, startByte + pos, encType, buffer, pos);
}
} catch (final SQLException e) {
logger.error("Error writing log part.", e);
throw new SQLException("Error writing log part", e);
} catch (final IOException e) {
logger.error("Error chunking.", e);
throw new SQLException("Error chunking", e);
}
} private void uploadLogPart(final DatabaseTransOperator transOperator, final int execId, final String name,
final int attempt, final int startByte, final int endByte, final EncodingType encType, final char[] buffer,
final int length) throws SQLException, IOException {
final String INSERT_EXECUTION_LOGS = "INSERT INTO execution_logs "
+ "(exec_id, name, attempt, enc_type, start_byte, end_byte, "
+ "log, upload_time) VALUES (?,?,?,?,?,?,?,?)"; byte[] buf = new String(buffer).getBytes();
if (encType == EncodingType.GZIP) {
buf = GZIPUtils.gzipBytes(buf, 0, buf.length);
} else if (length < buffer.length) {
buf = new String(buffer, 0, length).getBytes();
} transOperator.update(INSERT_EXECUTION_LOGS, execId, name, attempt, encType.getNumVal(), startByte,
startByte + length, buf, DateTime.now().getMillis());
以上代码直接放在ExecutionLogsDao内即可，修改引用的地方
至此日志显示中文乱码问题即可解决。以上为代码分析过程，操作过程可直接总结如下：
1.添加commons-lang3依赖
// https://mvnrepository.com/artifact/org.apache.commons/commons-lang3
compile group: 'org.apache.commons', name: 'commons-lang3', version: '3.4'
2.azkaban-web-server -->package azkaban.webapp.servlet-->ExecutorServlet-->将方法ajaxFetchExecFlowLogs和ajaxFetchJobLogs中的StringEscapeUtils.escapeHtml替换为org.apache.commons.lang3.StringEscapeUtils.escapeHtml3
3.azkaban-common-->package azkaban.utils-->FileIOUtils-->增加方法readUtf8FileEncode （文中第一段代码）
4.azkaban-common-->package azkaban.execapp-->FlowRunnerManager-->分别将readFlowLogs和readJobLogs中readUtf8File替换为readUtf8FileEncode
5.azkaban-common-->package azkaban.executor-->ExecutionLogsDao-->增加方法uploadLogFileEncode和uploadLogPart（重载）（文中第二段代码），将uploadLogFile（在70行左右的这个方法public void uploadLogFile(final int execId, final String name, final int attempt, final File... files)）中的uploadLogFile替换为uploadLogFileEncode
代码修改就是这些，我们可以发现代码中解析文件流都是采用UTF-8，如果生成的日志是其他编码就会出现乱码的情况，所以需要在读取文件的地方指定日志文件编码（还有一种思路就是Log4j写日志时候就指定编码，尝试没有成功就采用了现在的方式），来张前后对比图(对于已经写入数据库的日志，中文乱码问题无法解决，因为写入数据库前读取文件的方法不支持中文，修改代码之后入数据库的日志可以支持中文，另外代码中的GBK可以根据日志编码更换)
