Add partitioned sources for Parquet thrift / scrooge #1590

piyushnarang · 2016-08-30T20:40:25Z

Adding a couple of partitioned sources for Parquet thrift / scrooge similar to PartitionedDelimitedSource.
Usage is similar to the PartitionedDelimitedSource:

val myTypedPipe: TypedPipe[ (String, String), MyThriftObject] = ...
myTypedPipe.write(PartitionedParquetThriftSource[(String, String), MyThriftObject](path, "%s/%s"))
// writes out data to: path/%s/%s

Added a couple of unit tests and tested these sources out using a couple of test read / write jobs.

isnotinvain · 2016-08-30T20:58:06Z

Can this be slightly generalized such that any existing source can just "become" a partitioned source? Or maybe a factory that takes a Source and returns a PartitionedSource? or do we need to make one of these for each kind of source (parquet, scrooge, etc)

piyushnarang · 2016-08-30T21:33:18Z

Let me see if I can refactor / tweak things to generalize this. Gets a bit more complicated cause existing sources tend to pull in a set of traits with a specific type [T], when you trying including the PartitionedSchemed trait that pulls in an overlapping set of traits but with a different generic type [P,T]

piyushnarang · 2016-09-01T01:50:16Z

Poked around on this a bit to see if I could get it working. Seems like it's a bit more complicated / involved. Tried setting up a partitioned source which would delegate to another source but it seems like some of the methods on the Source class are protected so Scala's permissions seem to crib on that.
If possible, would be nice if we can tackle the generalization in a future iteration.

isnotinvain · 2016-09-01T02:35:30Z

...ge/src/main/scala/com/twitter/scalding/parquet/scrooge/PartitionedParquetScroogeSource.scala

+ * {{{
+ * val data = MyScroogeObject()
+ * IterablePipe(data, flowDef, mode)
+ *  .write(PartitionedParquetScroogeSource[(String, String), MyScroogeObject](path, "%s/%s"))


This isn't quite right is it? The partition key needs to be exposed right?
how about:

val data: TypedPipe[MyScroogeObject] = ??? data.map { obj => ((obj.country, obj.city), obj) }.write(PartitionedParquetScroogeSource[(String, String), MyScroogeObject](path, "%s/%s")

Yep, missed the map step there. Updated in the partitioned scrooge + thrift src docs.

piyushnarang · 2016-09-01T22:17:16Z

@isnotinvain - updated to clarify the docs to indicate that P must be a String / tuple of Strings. I think there's some code in PartitionSchemed (and some classes it delegates to) that expects the fields to be Strings if I'm understanding right - TemplatePartition

johnynek · 2016-09-09T22:56:55Z

...ge/src/main/scala/com/twitter/scalding/parquet/scrooge/PartitionedParquetScroogeSource.scala

+ *
+ */
+case class PartitionedParquetScroogeSource[P, T <: ThriftStruct](
+  path: String, template: String, fields: Fields = PartitionUtil.toFields(0, implicitly[TupleSetter[T]].arity))(implicit val mf: Manifest[T],


can we remove fields and Manifest[T]? Manifest is deprecated. And we really try to focus on the typed API now.

Hmm, seems like both were required based on the traits I was mixing in.
Manifest[T] as we're mixing in ParquetThriftBase.
Fields due to PartitionSchemed

can't we always fix given Fields to be PartitionUtil.toFields(0, implicitly[TupleSetter[T]].arity))? Also, can we fix ParquetThriftBase?

I don't want to confuse users with parameters they should really never change.

Yeah can set Fields to that. Can override it in PartitionedParquetScroogeSource and PartitionedParquetThriftSource unless you'd rather have it set in PartitionSchemed and not exposed to PartitionedParquet{Scrooge|Thrift}Source and PartitionedDelimitedSource?

Yeah, any new code should certainly not expose it. How much to remove/lock
down of old code is a question.
On Fri, Sep 9, 2016 at 14:55 Piyush Narang notifications@github.com wrote:

In
scalding-parquet-scrooge/src/main/scala/com/twitter/scalding/parquet/scrooge/PartitionedParquetScroogeSource.scala
#1590 (comment):

* val data: TypedPipe[MyScroogeObject] = ???

* data.map { obj =>

* ( (obj.country, obj.city), obj)

* }.write(PartitionedParquetScroogeSource(String, String), MyScroogeObject)

* }}}

* For reading it produces a pair (P, T) where P is the partition data, T is the corresponding

* scrooge object. Below is an example.

* {{{

* val in: TypedPipe[(String, String), MyScroogeObject] =

* TypedPipe.from( PartitionedParquetScroogeSource(String, String), MyScroogeObject )

* }}}

*/
+case class PartitionedParquetScroogeSource[P, T <: ThriftStruct](

path: String, template: String, fields: Fields = PartitionUtil.toFields(0, implicitly[TupleSetter[T]].arity))(implicit val mf: Manifest[T],

Yeah can set Fields to that. Can override it in
PartitionedParquetScroogeSource and PartitionedParquetThriftSource unless
you'd rather have it set in PartitionSchemed and not exposed to
PartitionedParquet{Scrooge|Thrift}Source and PartitionedDelimitedSource?

—
You are receiving this because you commented.

Reply to this email directly, view it on GitHub
https://github.com/twitter/scalding/pull/1590/files/363fed0f3820cb03a8faf8cf54a127b1f1f399f8#r78267367,
or mute the thread
https://github.com/notifications/unsubscribe-auth/AAEJdul73q_uAfVf2X0ZD2EJMHEO82N3ks5qof_vgaJpZM4Jw9Ah
.

johnynek

one minor comment

johnynek · 2016-09-15T20:20:12Z

scalding-parquet/src/main/scala/com/twitter/scalding/parquet/thrift/ParquetThrift.scala


-  def config: ParquetValueScheme.Config[T] = {
-    val config = new ParquetValueScheme.Config[T].withRecordClass(mf.runtimeClass.asInstanceOf[Class[T]])
+  def config(implicit ct: ClassTag[T]): ParquetValueScheme.Config[T] = {


why are we accepting ct as a parameter if we already have it on line 40?

Shall remove this, added this on one of my prior passes over the code when I was seeing some compile errors. Not needed anymore.

johnynek · 2016-09-15T20:24:24Z

scalding-parquet/src/main/scala/com/twitter/scalding/parquet/thrift/ParquetThrift.scala

-  def config: ParquetValueScheme.Config[T] = {
-    val config = new ParquetValueScheme.Config[T].withRecordClass(mf.runtimeClass.asInstanceOf[Class[T]])
+  def config(implicit ct: ClassTag[T]): ParquetValueScheme.Config[T] = {
+    val clazz = implicitly[ClassTag[T]].runtimeClass.asInstanceOf[Class[T]]


can't we just do val clazz = ct.runtimeClass.asInstanceOf[Class[T]] here

Hmm yeah, that's cleaner.

piyushnarang · 2016-09-15T23:06:26Z

@johnynek - updated based on your comments today. Have been looking at bit into trying out macros but the best I've got so far ended up being something on the lines of:

  def verifyTemplate(partitionFields: Fields, template: String): Unit = macro verifyTemplateImpl

  def verifyTemplateImpl(c: Context)(partitionFields: c.Expr[Fields], template: c.Expr[String]): c.Expr[Unit] = {
    import c.universe._

    reify {
            assert(
              partitionFields.splice.size() == "%s".r.findAllIn(template.splice).length,
              "Number of partition fields %s does not correspond to template (%s)".format(partitionFields.splice, template.splice))
    }
  }

This however ends up happening at runtime pretty much (and is identical to what we're doing today). Don't have a good sense of how to do this at compile time (tried a few things out but they seem to fail due to compile / misc errors).

johnynek · 2016-09-16T00:52:51Z

@piyushnarang here is a simple macro that checks that a string is a valid regex at compile time: https://gist.github.com/xuwei-k/4991805

that might give a hint.

We don't have to do it here, but I like this technique. We can punt on that if you like.

piyushnarang · 2016-09-16T02:00:04Z

Thanks @johnynek, that's a pretty useful link. I'd vote for following up with a separate PR for the macro bit. I can write up an issue so that we don't lose track of it and pick it up in the next week or so.

Piyush Narang added 4 commits August 30, 2016 10:46

Add partitioned parquet scrooge sources

e5dcc2d

Add partitioned parquet thrift source

057eb56

Rename sources

8863cd8

Make reformatter happy

4140d57

isnotinvain reviewed Sep 1, 2016
View reviewed changes

Piyush Narang added 3 commits September 1, 2016 11:04

Update usage documentation

a2d3468

Use scrooge reader for partition parquet scrooge test

5488d10

Tweak docs + remove unused imports

363fed0

johnynek reviewed Sep 9, 2016
View reviewed changes

Piyush Narang added 2 commits September 14, 2016 16:40

Remove fields param from partitioned parquet ctors

b02c281

Replace Manifests with ClassTags

09a8b75

johnynek requested changes Sep 15, 2016

View reviewed changes

johnynek reviewed Sep 15, 2016

View reviewed changes

Clean up implicit classtags

82d1387

johnynek approved these changes Sep 16, 2016

View reviewed changes

piyushnarang merged commit 0867f3e into twitter:develop Sep 16, 2016

piyushnarang mentioned this pull request Sep 16, 2016

Update template partition fields to format validation to a compile time macro #1598

Open

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add partitioned sources for Parquet thrift / scrooge #1590

Add partitioned sources for Parquet thrift / scrooge #1590

piyushnarang commented Aug 30, 2016

isnotinvain commented Aug 30, 2016

piyushnarang commented Aug 30, 2016

piyushnarang commented Sep 1, 2016

isnotinvain Sep 1, 2016

piyushnarang Sep 1, 2016

piyushnarang commented Sep 1, 2016

johnynek Sep 9, 2016

piyushnarang Sep 9, 2016

johnynek Sep 9, 2016

piyushnarang Sep 10, 2016

johnynek Sep 10, 2016

johnynek left a comment

johnynek Sep 15, 2016

piyushnarang Sep 15, 2016

johnynek Sep 15, 2016

piyushnarang Sep 15, 2016

piyushnarang commented Sep 15, 2016

johnynek commented Sep 16, 2016

piyushnarang commented Sep 16, 2016

Add partitioned sources for Parquet thrift / scrooge #1590

Add partitioned sources for Parquet thrift / scrooge #1590

Conversation

piyushnarang commented Aug 30, 2016

isnotinvain commented Aug 30, 2016

piyushnarang commented Aug 30, 2016

piyushnarang commented Sep 1, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

piyushnarang commented Sep 1, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

johnynek left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

piyushnarang commented Sep 15, 2016

johnynek commented Sep 16, 2016

piyushnarang commented Sep 16, 2016