仅仅是拷贝

waqulianjie · waqulianjie · commit 76234bebfddf · 2015-07-28T20:18:55.000+08:00
diff --git a/.classpath b/.classpath
@@ -0,0 +1,27 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<classpath>
+	<classpathentry kind="src" output="target/classes" path="src/main/java">
+		<attributes>
+			<attribute name="optional" value="true"/>
+			<attribute name="maven.pomderived" value="true"/>
+		</attributes>
+	</classpathentry>
+	<classpathentry kind="src" output="target/test-classes" path="src/test/java">
+		<attributes>
+			<attribute name="optional" value="true"/>
+			<attribute name="maven.pomderived" value="true"/>
+		</attributes>
+	</classpathentry>
+	<classpathentry kind="con" path="org.eclipse.m2e.MAVEN2_CLASSPATH_CONTAINER">
+		<attributes>
+			<attribute name="maven.pomderived" value="true"/>
+		</attributes>
+	</classpathentry>
+	<classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER/org.eclipse.jdt.internal.debug.ui.launcher.StandardVMType/jdk1.7.0_67"/>
+	<classpathentry kind="lib" path="lib/odps-sdk-commons-0.17.0-20150507.211034-68.jar"/>
+	<classpathentry kind="lib" path="lib/odps-sdk-core-0.17.0-20150507.211038-68.jar"/>
+	<classpathentry kind="lib" path="lib/aspectjrt-1.8.2.jar"/>
+	<classpathentry kind="lib" path="lib/javacsv-2.0.jar"/>
+	<classpathentry kind="lib" path="lib/snappy-java-1.1.1.6.jar"/>
+	<classpathentry kind="output" path="target/classes"/>
+</classpath>
diff --git a/.gitignore b/.gitignore
@@ -0,0 +1 @@
+/target/
diff --git a/.project b/.project
@@ -0,0 +1,23 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<projectDescription>
+	<name>odps_sink</name>
+	<comment></comment>
+	<projects>
+	</projects>
+	<buildSpec>
+		<buildCommand>
+			<name>org.eclipse.jdt.core.javabuilder</name>
+			<arguments>
+			</arguments>
+		</buildCommand>
+		<buildCommand>
+			<name>org.eclipse.m2e.core.maven2Builder</name>
+			<arguments>
+			</arguments>
+		</buildCommand>
+	</buildSpec>
+	<natures>
+		<nature>org.eclipse.jdt.core.javanature</nature>
+		<nature>org.eclipse.m2e.core.maven2Nature</nature>
+	</natures>
+</projectDescription>
diff --git a/.settings/org.eclipse.core.resources.prefs b/.settings/org.eclipse.core.resources.prefs
@@ -0,0 +1,4 @@
+eclipse.preferences.version=1
+encoding//src/main/java=UTF-8
+encoding//src/test/java=UTF-8
+encoding/<project>=UTF-8
diff --git a/.settings/org.eclipse.jdt.core.prefs b/.settings/org.eclipse.jdt.core.prefs
@@ -0,0 +1,13 @@
+eclipse.preferences.version=1
+org.eclipse.jdt.core.compiler.codegen.inlineJsrBytecode=enabled
+org.eclipse.jdt.core.compiler.codegen.methodParameters=do not generate
+org.eclipse.jdt.core.compiler.codegen.targetPlatform=1.7
+org.eclipse.jdt.core.compiler.codegen.unusedLocal=preserve
+org.eclipse.jdt.core.compiler.compliance=1.7
+org.eclipse.jdt.core.compiler.debug.lineNumber=generate
+org.eclipse.jdt.core.compiler.debug.localVariable=generate
+org.eclipse.jdt.core.compiler.debug.sourceFile=generate
+org.eclipse.jdt.core.compiler.problem.assertIdentifier=error
+org.eclipse.jdt.core.compiler.problem.enumIdentifier=error
+org.eclipse.jdt.core.compiler.problem.forbiddenReference=warning
+org.eclipse.jdt.core.compiler.source=1.7
diff --git a/.settings/org.eclipse.m2e.core.prefs b/.settings/org.eclipse.m2e.core.prefs
@@ -0,0 +1,4 @@
+activeProfiles=
+eclipse.preferences.version=1
+resolveWorkspaceProjects=true
+version=1
diff --git a/README.md b/README.md
@@ -1,2 +1,8 @@
 # odps_sink
 odps_sink for flume
+
+odps is a product of aliyun
+
+by shaolianjie waqu
+
+
diff --git a/lib/aspectjrt-1.8.2.jar b/lib/aspectjrt-1.8.2.jar
diff --git a/lib/javacsv-2.0.jar b/lib/javacsv-2.0.jar
diff --git a/lib/odps-sdk-commons-0.17.0-20150507.211034-68.jar b/lib/odps-sdk-commons-0.17.0-20150507.211034-68.jar
diff --git a/lib/odps-sdk-core-0.17.0-20150507.211038-68.jar b/lib/odps-sdk-core-0.17.0-20150507.211038-68.jar
diff --git a/lib/snappy-java-1.1.1.6.jar b/lib/snappy-java-1.1.1.6.jar
diff --git a/pom.xml b/pom.xml
@@ -0,0 +1,25 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+
+	<groupId>com.waqu</groupId>
+	<artifactId>sink</artifactId>
+	<version>0.0.1-SNAPSHOT</version>
+	<packaging>jar</packaging>
+
+	<name>sink</name>
+	<url>http://maven.apache.org</url>
+
+	<properties>
+		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+	</properties>
+
+	<dependencies>
+		<dependency>
+			<groupId>org.apache.flume</groupId>
+			<artifactId>flume-ng-core</artifactId>
+			<version>1.6.0</version>
+		</dependency>
+
+	</dependencies>
+</project>
diff --git a/src/main/java/com/waqu/sink/OdpsSink.java b/src/main/java/com/waqu/sink/OdpsSink.java
@@ -0,0 +1,224 @@
+package com.waqu.sink;
+
+import java.text.DateFormat;
+import java.text.SimpleDateFormat;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map.Entry;
+import java.util.Random;
+import java.util.Set;
+
+import org.apache.commons.lang.StringUtils;
+import org.apache.flume.Channel;
+import org.apache.flume.Context;
+import org.apache.flume.Event;
+import org.apache.flume.EventDeliveryException;
+import org.apache.flume.Transaction;
+import org.apache.flume.conf.Configurable;
+import org.apache.flume.sink.AbstractSink;
+
+import com.aliyun.odps.Column;
+import com.aliyun.odps.Odps;
+import com.aliyun.odps.OdpsException;
+import com.aliyun.odps.Partition;
+import com.aliyun.odps.PartitionSpec;
+import com.aliyun.odps.Table;
+import com.aliyun.odps.TableSchema;
+import com.aliyun.odps.account.AliyunAccount;
+import com.aliyun.odps.data.ArrayRecord;
+import com.aliyun.odps.data.Record;
+import com.aliyun.odps.tunnel.StreamClient;
+import com.aliyun.odps.tunnel.StreamClient.ShardState;
+import com.aliyun.odps.tunnel.TableTunnel;
+import com.aliyun.odps.tunnel.TunnelException;
+import com.aliyun.odps.tunnel.io.StreamRecordPack;
+import com.aliyun.odps.tunnel.io.StreamWriter;
+import com.google.common.collect.Lists;
+
+public class OdpsSink extends AbstractSink implements Configurable {
+
+	private TableTunnel tableTunnel;
+	private Table table;
+	private String currentPartition = "";
+	private final DateFormat totalDf = new SimpleDateFormat("yyyy-MM-dd-HH");
+	private final DateFormat dateDf = new SimpleDateFormat("yyyy-MM-dd");
+	private final DateFormat hourDf = new SimpleDateFormat("HH");
+	private int bathSize = 10;
+	private StreamWriter writers[];
+	private StreamClient streamClient;
+	private PartitionSpec partitionsSpec;
+	private Column[] columns;
+	private Random random = new Random();
+
+	@Override
+	public void configure(Context context) {
+
+		String accessId = context.getString("sink.odps.access_id", "####");
+		String accessKey = context.getString("sink.odps.access_key", "######");
+		String endPoint = context.getString("sink.odps.end_point", "http://odps-ext.aliyun-inc.com/api");
+		String tunnelEndPoint = context.getString("sink.tunnel.end_point", "http://dh-ext.odps.aliyun-inc.com");
+		String tableName = context.getString("sink.table", "####");
+		String project = context.getString("sink.project", "######");
+
+		Odps odps = new Odps(new AliyunAccount(accessId, accessKey));
+		odps.setEndpoint(endPoint);
+		odps.setDefaultProject(project);
+		tableTunnel = new TableTunnel(odps);
+		tableTunnel.setEndpoint(tunnelEndPoint);
+
+		try {
+			streamClient = tableTunnel.createStreamClient(project, tableName);
+			streamClient.loadShard(5);
+		} catch (TunnelException e) {
+			throw new RuntimeException("createStreamClient create error", e);
+		}
+
+		table = odps.tables().get(tableName);
+
+		bathSize = context.getInteger("sink.batchSize", 10);
+		
+		TableSchema tableSchema = streamClient.getStreamSchema();
+		columns = (Column[]) tableSchema.getColumns().toArray(new Column[tableSchema.getColumns().size()]);
+
+	}
+
+	private Set<String> getAllPartitions() {
+		Set<String> partitionSet = new HashSet<String>();
+
+		List<Partition> list = table.getPartitions();
+		for (Partition partition : list) {
+			PartitionSpec ps = partition.getPartitionSpec();
+			String ctime = ps.get("ctime");
+			String hour = ps.get("hour");
+			partitionSet.add(ctime + "-" + hour);
+		}
+		return partitionSet;
+	}
+
+	private void addPartition(String ctime, String hour) throws OdpsException {
+		table.createPartition(genPartitionSpec(ctime, hour));
+	}
+
+	private PartitionSpec genPartitionSpec(String ctime, String hour) {
+		PartitionSpec ps = new PartitionSpec();
+		ps.set("ctime", ctime);
+		ps.set("hour", hour);
+		return ps;
+	}
+
+	private StreamWriter[] getWriters(String serverTime) throws Exception {
+		long now = Long.parseLong(serverTime);
+		Date date = new Date();
+		date.setTime(now);
+		String currentPar = totalDf.format(date);
+		if (writers == null || writers.length==0 || !StringUtils.equals(currentPar, currentPartition)) {
+			String ctime = dateDf.format(date);
+			String hour = hourDf.format(date);
+			System.out.println("init new stream writers for hour:" + hour);
+			Set<String> partitionSet = getAllPartitions();
+			if (!partitionSet.contains(currentPar)) {
+				addPartition(ctime, hour);
+			}
+			freshPartitionSpec(ctime, hour);
+			buildStreamWriter();
+			currentPartition = currentPar;
+		}
+		return writers;
+	}
+
+	private void freshPartitionSpec(String ctime, String hour) {
+		this.partitionsSpec = genPartitionSpec(ctime, hour);
+		
+	}
+
+	private void buildStreamWriter() throws Exception {
+		HashMap<Long, ShardState> shardMap = streamClient.getShardStatus();
+		List<Long> shardIdList = new ArrayList<Long>();
+		int i = 0;
+		for(Entry<Long, ShardState> entry:shardMap.entrySet()){
+			if(StringUtils.equals(entry.getValue().name(), StreamClient.ShardState.LOADED.name())){
+				shardIdList.add(entry.getKey());
+				i++;
+			}
+			if(i>=10){
+				break;
+			}
+		}
+		
+		if (shardIdList.size() > 0) {
+			writers = new StreamWriter[shardIdList.size()];
+			for (int j = 0; j < shardIdList.size() ; j++)
+				writers[j] = streamClient.openStreamWriter(((Long) shardIdList.get(j)).longValue());
+
+		} else {
+			throw new RuntimeException("OdpsUploadProcessor buildStreamWriter() error, have not loaded shards.");
+		}
+	}
+
+	@Override
+	public void start() {
+	}
+
+	@Override
+	public void stop() {
+	}
+
+	@Override
+	public Status process() throws EventDeliveryException {
+		Status status = null;
+		// Start transaction
+		Channel ch = getChannel();
+		Transaction txn = ch.getTransaction();
+		txn.begin();
+		try {
+			StreamWriter writers[] = null;
+			List<Record> recordList = Lists.newArrayList();
+			for (int i = 0; i < bathSize; i++) {
+				Event event = ch.take();
+				String data = new String(event.getBody(), "UTF-8");
+				if (StringUtils.isNotEmpty(data)) {
+					String[] lines = StringUtils.split(data, "\n");
+//					System.out.println("data:"+data);
+					for (String line : lines) {
+						String[] cols = StringUtils.splitPreserveAllTokens(line, "\t");
+						if (cols.length < 7) {
+//							System.out.println("data size error now:"+cols.length+" line:"+line);
+							continue;
+						}
+						if (recordList.size() == 0) {
+							writers = this.getWriters(cols[1]);
+//							System.out.println("writers.length:"+writers.length);
+							// ntype\tserver_time\t ..
+						}
+						Record record =  new ArrayRecord(columns);
+						for (int k = 0; k < columns.length; k++) {
+							record.setString(k, cols[k]);
+						}
+						recordList.add(record);
+					}
+				}
+			}
+			StreamRecordPack recordPack = new StreamRecordPack(streamClient.getStreamSchema());
+			for(Record record:recordList){
+				recordPack.append(record);
+			}
+//			System.out.println("recordPack.getRecordCount:"+recordPack.getRecordCount());
+//			System.out.println("partitionsSpec.toString:"+partitionsSpec.toString());
+			writers[random.nextInt(writers.length)].write(partitionsSpec, recordPack);
+			txn.commit();
+			status = Status.READY;
+		} catch (Throwable t) {
+			txn.rollback();
+			status = Status.BACKOFF;
+			if (t instanceof Error) {
+				throw (Error) t;
+			}
+		} finally {
+			txn.close();
+		}
+		return status;
+	}
+}
diff --git a/使用手册 b/使用手册
@@ -0,0 +1,15 @@
+1.打个jar
+2.放到flume的lib下，并且把odps依赖的jar都放进去(就是lib文件夹下的东西)
+3.修改配置文件
+
+注1:lib下的是阿里云odps的jar包，丫的没开源，maven地址都没有，至少当初我用的时候是这样的，跟丫拼了
+注2:目前代码只支持 天级+小时级分区  ctime天级分区，hour小时级分区，也是醉了，这种普通的需要 连个sdk都没有，也是醉了
+注3:先把成日志的表改成hub表，都他妈是坑，文档的东西啥都没有，也没有人管一管！！！
+
+TODO:待升级的东西
+1.具体时间区分 可配置  我觉得至少应该支持到分钟级的分区吧，分钟级的报表还是很重要的，百度可以做到15分钟
+2.导入的hub表字段可配置，以及字段类型
+3.目前出入的一行数据，是用\n分隔的一个大字符串，我这边进行了一次手动处理，用\n分隔之后，把每条数据放到odps里边，这个根据具体业务，提取一些出来
+
+ 
+迁移odps的过程，史上最2，没有之一
