{"action":"created","comment":{"url":"https://api.github.com/repos/linkedin/gobblin/pulls/comments/28466742","id":28466742,"diff_hunk":"@@ -0,0 +1,284 @@\n+/* (c) 2014 LinkedIn Corp. All rights reserved.\n+ *\n+ * Licensed under the Apache License, Version 2.0 (the \"License\"); you may not use\n+ * this file except in compliance with the License. You may obtain a copy of the\n+ * License at  http://www.apache.org/licenses/LICENSE-2.0\n+ *\n+ * Unless required by applicable law or agreed to in writing, software distributed\n+ * under the License is distributed on an \"AS IS\" BASIS, WITHOUT WARRANTIES OR\n+ * CONDITIONS OF ANY KIND, either express or implied.\n+ */\n+\n+package gobblin.writer;\n+\n+import java.io.IOException;\n+import java.util.Map;\n+\n+import org.apache.avro.Schema;\n+import org.apache.avro.generic.GenericRecord;\n+import org.apache.hadoop.fs.Path;\n+import org.joda.time.DateTimeZone;\n+import org.joda.time.format.DateTimeFormat;\n+import org.joda.time.format.DateTimeFormatter;\n+import org.slf4j.Logger;\n+import org.slf4j.LoggerFactory;\n+\n+import com.google.common.base.Optional;\n+import com.google.common.base.Preconditions;\n+import com.google.common.collect.Maps;\n+\n+import gobblin.configuration.ConfigurationKeys;\n+import gobblin.configuration.State;\n+import gobblin.util.AvroUtils;\n+import gobblin.util.ForkOperatorUtils;\n+\n+\n+/**\n+ * Implementation of {@link DataWriter} that writes data into a date-partitioned directory structure based on the value\n+ * of a specific field in each Avro record.\n+ *\n+ * <p>\n+ *\n+ * The writer takes in each Avro record, extracts out a specific field, and based on the the value of the column it\n+ * writes the data into a specific file. The column is assumed to be of type {@link Long}, and the value is treated as\n+ * a timestamp. By default, the timestamp is converted into a directory of the form [yyyy]/[MM][dd]/. However, the\n+ * actual pattern used to construct the path is specified by {@link ConfigurationKeys#WRITER_PARTITION_PATTERN}. This\n+ * class uses Joda's {@link DateTimeFormatter} to parse a timestamp, and convert it to a {@link String}. The\n+ * {@link String} is then converted to a Path, which specifies where the record will be written.\n+ *\n+ * <p>\n+ *\n+ * By default, The complete output path for a directory is \"/baseFilePath/daily/[yyyy]/[MM]/[dd]/fileName.avro\". The\n+ * baseFilePath is specified by the configuration key {@link ConfigurationKeys#WRITER_FILE_PATH}. The writer uses the\n+ * configuration key {@link ConfigurationKeys#WRITER_PARTITION_COLUMN_NAME} to determine the name of the column to\n+ * partition by. The \"daily\" portion of the path is also configurable, and can be changed by setting\n+ * {@link ConfigurationKeys#DEFAULT_WRITER_PARTITION_LEVEL}.\n+ *\n+ * <p>\n+ *\n+ * The class works by maintaining a {@link Map} of {@link Path}s to {@link DataWriter}s called {@link #pathToWriterMap}.\n+ * The {@link AvroHdfsTimePartitionedWriter#write(GenericRecord)} takes each record, extracts out the the value of the\n+ * field specified by {@link ConfigurationKeys#WRITER_PARTITION_COLUMN_NAME}, and constructs a {@link Path} based on the\n+ * field value. If a {@link DataWriter} already exists for the path, the record is simply written to the existing\n+ * {@link DataWriter}. If one does not exist, then a new {@link DataWriter} is created, added to the Map, and then the\n+ * input record is written to the writer.\n+ *\n+ * <p>\n+ *\n+ * By using the above approach methods such as {@link DataWriter#commit()}, {@link DataWriter#close()}, etc. are simple.\n+ * The implementation is to simply iterate over the values of {@link #pathToWriterMap} and call the corresponding method\n+ * on each {@link DataWriter}.\n+ */\n+public class AvroHdfsTimePartitionedWriter implements DataWriter<GenericRecord> {\n+\n+  /**\n+   * This is the base file path that all data will be written to. By default, data will be written to\n+   * /baseFilePath/daily/[yyyy]/[MM]/[dd]/.\n+   */\n+  private final String baseFilePath;\n+\n+  /**\n+   * The name of the column that the writer will use to partition the data.\n+   */\n+  private final String partitionColumnName;\n+\n+  /**\n+   * The name that separates the {@link #baseFilePath} from the path created by the {@link #timestampToPathFormatter}.\n+   * The default value is specified by {@link ConfigurationKeys#DEFAULT_WRITER_PARTITION_LEVEL}.\n+   */\n+  private final String partitionLevel;\n+\n+  /**\n+   * This {@link DateTimeFormatter} controls how the date-partitioned directories will be created. The default pattern\n+   * used is specified by {@link ConfigurationKeys#DEFAULT_WRITER_PARTITION_PATTERN}. The\n+   * {@link DateTimeFormatter#print(long)} method is used to convert a timestamp to a {@link String}.\n+   */\n+  private final DateTimeFormatter timestampToPathFormatter;\n+\n+  /**\n+   * Maps a {@link Path} to the the {@link DataWriter} that is writing data to the Path.\n+   */\n+  private final Map<Path, DataWriter<GenericRecord>> pathToWriterMap = Maps.newHashMap();\n+\n+  // Variables needed to build DataWriters\n+  private final Destination destination;\n+  private final String writerId;\n+  private final Schema schema;\n+  private final WriterOutputFormat writerOutputFormat;\n+  private final State properties;\n+  private final int branch;\n+\n+  private static final Logger LOG = LoggerFactory.getLogger(AvroHdfsTimePartitionedWriter.class);\n+\n+  public AvroHdfsTimePartitionedWriter(Destination destination, String writerId, Schema schema,\n+      WriterOutputFormat writerOutputFormat, int branch) {\n+\n+    this.destination = destination;\n+    this.writerId = writerId;\n+    this.schema = schema;\n+    this.writerOutputFormat = writerOutputFormat;\n+    this.branch = branch;\n+    this.properties = destination.getProperties();\n+\n+    // Confirm that all input parameters are not null\n+    Preconditions.checkNotNull(this.destination);\n+    Preconditions.checkNotNull(this.writerId);\n+    Preconditions.checkNotNull(this.schema);\n+    Preconditions.checkNotNull(this.writerOutputFormat);\n+    Preconditions.checkNotNull(this.branch);\n+    Preconditions.checkNotNull(this.properties);\n+\n+    this.baseFilePath = this.properties.getProp(getWriterFilePath(this.branch));","path":"gobblin-core/src/main/java/gobblin/writer/AvroHdfsTimePartitionedWriter.java","position":131,"original_position":131,"commit_id":"7a434d7541639f75e1e6eaec7078e25630944d45","original_commit_id":"7a434d7541639f75e1e6eaec7078e25630944d45","user":{"login":"liyinan926","id":3709137,"avatar_url":"https://avatars.githubusercontent.com/u/3709137?v=3","gravatar_id":"","url":"https://api.github.com/users/liyinan926","html_url":"https://github.com/liyinan926","followers_url":"https://api.github.com/users/liyinan926/followers","following_url":"https://api.github.com/users/liyinan926/following{/other_user}","gists_url":"https://api.github.com/users/liyinan926/gists{/gist_id}","starred_url":"https://api.github.com/users/liyinan926/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/liyinan926/subscriptions","organizations_url":"https://api.github.com/users/liyinan926/orgs","repos_url":"https://api.github.com/users/liyinan926/repos","events_url":"https://api.github.com/users/liyinan926/events{/privacy}","received_events_url":"https://api.github.com/users/liyinan926/received_events","type":"User","site_admin":false},"body":"Better to check if properties contains the key.","created_at":"2015-04-15T21:50:53Z","updated_at":"2015-04-15T21:50:53Z","html_url":"https://github.com/linkedin/gobblin/pull/86#discussion_r28466742","pull_request_url":"https://api.github.com/repos/linkedin/gobblin/pulls/86","_links":{"self":{"href":"https://api.github.com/repos/linkedin/gobblin/pulls/comments/28466742"},"html":{"href":"https://github.com/linkedin/gobblin/pull/86#discussion_r28466742"},"pull_request":{"href":"https://api.github.com/repos/linkedin/gobblin/pulls/86"}}},"pull_request":{"url":"https://api.github.com/repos/linkedin/gobblin/pulls/86","id":33113235,"html_url":"https://github.com/linkedin/gobblin/pull/86","diff_url":"https://github.com/linkedin/gobblin/pull/86.diff","patch_url":"https://github.com/linkedin/gobblin/pull/86.patch","issue_url":"https://api.github.com/repos/linkedin/gobblin/issues/86","number":86,"state":"open","locked":false,"title":"Date partition writer","user":{"login":"sahilTakiar","id":1282583,"avatar_url":"https://avatars.githubusercontent.com/u/1282583?v=3","gravatar_id":"","url":"https://api.github.com/users/sahilTakiar","html_url":"https://github.com/sahilTakiar","followers_url":"https://api.github.com/users/sahilTakiar/followers","following_url":"https://api.github.com/users/sahilTakiar/following{/other_user}","gists_url":"https://api.github.com/users/sahilTakiar/gists{/gist_id}","starred_url":"https://api.github.com/users/sahilTakiar/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/sahilTakiar/subscriptions","organizations_url":"https://api.github.com/users/sahilTakiar/orgs","repos_url":"https://api.github.com/users/sahilTakiar/repos","events_url":"https://api.github.com/users/sahilTakiar/events{/privacy}","received_events_url":"https://api.github.com/users/sahilTakiar/received_events","type":"User","site_admin":false},"body":"This pull requests is for adding a Writer that can write date-partitioned data. There is a new AvroDataWriterBuilder called AvroDatePartitionedWriterBuilder, which creates a AvroHdfsDatePartitionedWriter. Also, wrote a test for AvroHdfsDatePartitionedWriter.\r\n\r\nThe AvroHdfsDatePartitionedWriter contains a Map from Path to AvroHdfsDataWriters. More explanation is in the JavaDocs.\r\n\r\nMinor cleanup to BaseDataPublisher, more javadocs, code formatting, etc.","created_at":"2015-04-12T01:44:09Z","updated_at":"2015-04-15T21:50:53Z","closed_at":null,"merged_at":null,"merge_commit_sha":"1de17acb060d9ee8c32cc8d2ef99dca2389b02db","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/linkedin/gobblin/pulls/86/commits","review_comments_url":"https://api.github.com/repos/linkedin/gobblin/pulls/86/comments","review_comment_url":"https://api.github.com/repos/linkedin/gobblin/pulls/comments{/number}","comments_url":"https://api.github.com/repos/linkedin/gobblin/issues/86/comments","statuses_url":"https://api.github.com/repos/linkedin/gobblin/statuses/7a434d7541639f75e1e6eaec7078e25630944d45","head":{"label":"sahilTakiar:date-partition-writer","ref":"date-partition-writer","sha":"7a434d7541639f75e1e6eaec7078e25630944d45","user":{"login":"sahilTakiar","id":1282583,"avatar_url":"https://avatars.githubusercontent.com/u/1282583?v=3","gravatar_id":"","url":"https://api.github.com/users/sahilTakiar","html_url":"https://github.com/sahilTakiar","followers_url":"https://api.github.com/users/sahilTakiar/followers","following_url":"https://api.github.com/users/sahilTakiar/following{/other_user}","gists_url":"https://api.github.com/users/sahilTakiar/gists{/gist_id}","starred_url":"https://api.github.com/users/sahilTakiar/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/sahilTakiar/subscriptions","organizations_url":"https://api.github.com/users/sahilTakiar/orgs","repos_url":"https://api.github.com/users/sahilTakiar/repos","events_url":"https://api.github.com/users/sahilTakiar/events{/privacy}","received_events_url":"https://api.github.com/users/sahilTakiar/received_events","type":"User","site_admin":false},"repo":{"id":31343048,"name":"gobblin","full_name":"sahilTakiar/gobblin","owner":{"login":"sahilTakiar","id":1282583,"avatar_url":"https://avatars.githubusercontent.com/u/1282583?v=3","gravatar_id":"","url":"https://api.github.com/users/sahilTakiar","html_url":"https://github.com/sahilTakiar","followers_url":"https://api.github.com/users/sahilTakiar/followers","following_url":"https://api.github.com/users/sahilTakiar/following{/other_user}","gists_url":"https://api.github.com/users/sahilTakiar/gists{/gist_id}","starred_url":"https://api.github.com/users/sahilTakiar/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/sahilTakiar/subscriptions","organizations_url":"https://api.github.com/users/sahilTakiar/orgs","repos_url":"https://api.github.com/users/sahilTakiar/repos","events_url":"https://api.github.com/users/sahilTakiar/events{/privacy}","received_events_url":"https://api.github.com/users/sahilTakiar/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/sahilTakiar/gobblin","description":"","fork":true,"url":"https://api.github.com/repos/sahilTakiar/gobblin","forks_url":"https://api.github.com/repos/sahilTakiar/gobblin/forks","keys_url":"https://api.github.com/repos/sahilTakiar/gobblin/keys{/key_id}","collaborators_url":"https://api.github.com/repos/sahilTakiar/gobblin/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/sahilTakiar/gobblin/teams","hooks_url":"https://api.github.com/repos/sahilTakiar/gobblin/hooks","issue_events_url":"https://api.github.com/repos/sahilTakiar/gobblin/issues/events{/number}","events_url":"https://api.github.com/repos/sahilTakiar/gobblin/events","assignees_url":"https://api.github.com/repos/sahilTakiar/gobblin/assignees{/user}","branches_url":"https://api.github.com/repos/sahilTakiar/gobblin/branches{/branch}","tags_url":"https://api.github.com/repos/sahilTakiar/gobblin/tags","blobs_url":"https://api.github.com/repos/sahilTakiar/gobblin/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/sahilTakiar/gobblin/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/sahilTakiar/gobblin/git/refs{/sha}","trees_url":"https://api.github.com/repos/sahilTakiar/gobblin/git/trees{/sha}","statuses_url":"https://api.github.com/repos/sahilTakiar/gobblin/statuses/{sha}","languages_url":"https://api.github.com/repos/sahilTakiar/gobblin/languages","stargazers_url":"https://api.github.com/repos/sahilTakiar/gobblin/stargazers","contributors_url":"https://api.github.com/repos/sahilTakiar/gobblin/contributors","subscribers_url":"https://api.github.com/repos/sahilTakiar/gobblin/subscribers","subscription_url":"https://api.github.com/repos/sahilTakiar/gobblin/subscription","commits_url":"https://api.github.com/repos/sahilTakiar/gobblin/commits{/sha}","git_commits_url":"https://api.github.com/repos/sahilTakiar/gobblin/git/commits{/sha}","comments_url":"https://api.github.com/repos/sahilTakiar/gobblin/comments{/number}","issue_comment_url":"https://api.github.com/repos/sahilTakiar/gobblin/issues/comments{/number}","contents_url":"https://api.github.com/repos/sahilTakiar/gobblin/contents/{+path}","compare_url":"https://api.github.com/repos/sahilTakiar/gobblin/compare/{base}...{head}","merges_url":"https://api.github.com/repos/sahilTakiar/gobblin/merges","archive_url":"https://api.github.com/repos/sahilTakiar/gobblin/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/sahilTakiar/gobblin/downloads","issues_url":"https://api.github.com/repos/sahilTakiar/gobblin/issues{/number}","pulls_url":"https://api.github.com/repos/sahilTakiar/gobblin/pulls{/number}","milestones_url":"https://api.github.com/repos/sahilTakiar/gobblin/milestones{/number}","notifications_url":"https://api.github.com/repos/sahilTakiar/gobblin/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/sahilTakiar/gobblin/labels{/name}","releases_url":"https://api.github.com/repos/sahilTakiar/gobblin/releases{/id}","created_at":"2015-02-26T00:28:16Z","updated_at":"2015-04-15T20:53:12Z","pushed_at":"2015-04-15T20:54:28Z","git_url":"git://github.com/sahilTakiar/gobblin.git","ssh_url":"git@github.com:sahilTakiar/gobblin.git","clone_url":"https://github.com/sahilTakiar/gobblin.git","svn_url":"https://github.com/sahilTakiar/gobblin","homepage":null,"size":76478,"stargazers_count":0,"watchers_count":0,"language":"Java","has_issues":false,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":0,"mirror_url":null,"open_issues_count":0,"forks":0,"open_issues":0,"watchers":0,"default_branch":"master"}},"base":{"label":"linkedin:master","ref":"master","sha":"41e9f5718501378e2aa816633a7068bd83793d23","user":{"login":"linkedin","id":357098,"avatar_url":"https://avatars.githubusercontent.com/u/357098?v=3","gravatar_id":"","url":"https://api.github.com/users/linkedin","html_url":"https://github.com/linkedin","followers_url":"https://api.github.com/users/linkedin/followers","following_url":"https://api.github.com/users/linkedin/following{/other_user}","gists_url":"https://api.github.com/users/linkedin/gists{/gist_id}","starred_url":"https://api.github.com/users/linkedin/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/linkedin/subscriptions","organizations_url":"https://api.github.com/users/linkedin/orgs","repos_url":"https://api.github.com/users/linkedin/repos","events_url":"https://api.github.com/users/linkedin/events{/privacy}","received_events_url":"https://api.github.com/users/linkedin/received_events","type":"Organization","site_admin":false},"repo":{"id":27392543,"name":"gobblin","full_name":"linkedin/gobblin","owner":{"login":"linkedin","id":357098,"avatar_url":"https://avatars.githubusercontent.com/u/357098?v=3","gravatar_id":"","url":"https://api.github.com/users/linkedin","html_url":"https://github.com/linkedin","followers_url":"https://api.github.com/users/linkedin/followers","following_url":"https://api.github.com/users/linkedin/following{/other_user}","gists_url":"https://api.github.com/users/linkedin/gists{/gist_id}","starred_url":"https://api.github.com/users/linkedin/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/linkedin/subscriptions","organizations_url":"https://api.github.com/users/linkedin/orgs","repos_url":"https://api.github.com/users/linkedin/repos","events_url":"https://api.github.com/users/linkedin/events{/privacy}","received_events_url":"https://api.github.com/users/linkedin/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/linkedin/gobblin","description":"","fork":false,"url":"https://api.github.com/repos/linkedin/gobblin","forks_url":"https://api.github.com/repos/linkedin/gobblin/forks","keys_url":"https://api.github.com/repos/linkedin/gobblin/keys{/key_id}","collaborators_url":"https://api.github.com/repos/linkedin/gobblin/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/linkedin/gobblin/teams","hooks_url":"https://api.github.com/repos/linkedin/gobblin/hooks","issue_events_url":"https://api.github.com/repos/linkedin/gobblin/issues/events{/number}","events_url":"https://api.github.com/repos/linkedin/gobblin/events","assignees_url":"https://api.github.com/repos/linkedin/gobblin/assignees{/user}","branches_url":"https://api.github.com/repos/linkedin/gobblin/branches{/branch}","tags_url":"https://api.github.com/repos/linkedin/gobblin/tags","blobs_url":"https://api.github.com/repos/linkedin/gobblin/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/linkedin/gobblin/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/linkedin/gobblin/git/refs{/sha}","trees_url":"https://api.github.com/repos/linkedin/gobblin/git/trees{/sha}","statuses_url":"https://api.github.com/repos/linkedin/gobblin/statuses/{sha}","languages_url":"https://api.github.com/repos/linkedin/gobblin/languages","stargazers_url":"https://api.github.com/repos/linkedin/gobblin/stargazers","contributors_url":"https://api.github.com/repos/linkedin/gobblin/contributors","subscribers_url":"https://api.github.com/repos/linkedin/gobblin/subscribers","subscription_url":"https://api.github.com/repos/linkedin/gobblin/subscription","commits_url":"https://api.github.com/repos/linkedin/gobblin/commits{/sha}","git_commits_url":"https://api.github.com/repos/linkedin/gobblin/git/commits{/sha}","comments_url":"https://api.github.com/repos/linkedin/gobblin/comments{/number}","issue_comment_url":"https://api.github.com/repos/linkedin/gobblin/issues/comments{/number}","contents_url":"https://api.github.com/repos/linkedin/gobblin/contents/{+path}","compare_url":"https://api.github.com/repos/linkedin/gobblin/compare/{base}...{head}","merges_url":"https://api.github.com/repos/linkedin/gobblin/merges","archive_url":"https://api.github.com/repos/linkedin/gobblin/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/linkedin/gobblin/downloads","issues_url":"https://api.github.com/repos/linkedin/gobblin/issues{/number}","pulls_url":"https://api.github.com/repos/linkedin/gobblin/pulls{/number}","milestones_url":"https://api.github.com/repos/linkedin/gobblin/milestones{/number}","notifications_url":"https://api.github.com/repos/linkedin/gobblin/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/linkedin/gobblin/labels{/name}","releases_url":"https://api.github.com/repos/linkedin/gobblin/releases{/id}","created_at":"2014-12-01T18:10:50Z","updated_at":"2015-04-15T21:03:48Z","pushed_at":"2015-04-15T21:35:19Z","git_url":"git://github.com/linkedin/gobblin.git","ssh_url":"git@github.com:linkedin/gobblin.git","clone_url":"https://github.com/linkedin/gobblin.git","svn_url":"https://github.com/linkedin/gobblin","homepage":null,"size":155193,"stargazers_count":33,"watchers_count":33,"language":"Java","has_issues":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":17,"mirror_url":null,"open_issues_count":6,"forks":17,"open_issues":6,"watchers":33,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/linkedin/gobblin/pulls/86"},"html":{"href":"https://github.com/linkedin/gobblin/pull/86"},"issue":{"href":"https://api.github.com/repos/linkedin/gobblin/issues/86"},"comments":{"href":"https://api.github.com/repos/linkedin/gobblin/issues/86/comments"},"review_comments":{"href":"https://api.github.com/repos/linkedin/gobblin/pulls/86/comments"},"review_comment":{"href":"https://api.github.com/repos/linkedin/gobblin/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/linkedin/gobblin/pulls/86/commits"},"statuses":{"href":"https://api.github.com/repos/linkedin/gobblin/statuses/7a434d7541639f75e1e6eaec7078e25630944d45"}}}}