Packages

•  package root

  Definition Classes

  root

•  package com

  Definition Classes

  root

•  package spotify

  Definition Classes

  com

•  package scio

  Definition Classes

  spotify

•  package extra

  Definition Classes

  scio

•  package sparkey

  Main package for Sparkey side input APIs.

  Main package for Sparkey side input APIs. Import all.

  import com.spotify.scio.extra.sparkey._

  To save an SCollection[(String, String)] to a Sparkey fileset:

  val s = sc.parallelize(Seq("a" -> "one", "b" -> "two"))
  s.saveAsSparkey("gs://<bucket>/<path>/<sparkey-prefix>")
  
  // with multiple shards, sharded by MurmurHash3 of the key
  s.saveAsSparkey("gs://<bucket>/<path>/<sparkey-dir>", numShards=2)

  A previously-saved sparkey can be loaded as a side input:

  sc.sparkeySideInput("gs://<bucket>/<path>/<sparkey-prefix>")

  A sharded collection of Sparkey files can also be used as a side input by specifying a glob path:

  sc.sparkeySideInput("gs://<bucket>/<path>/<sparkey-dir>/part-*")

  When the sparkey is needed only temporarily, the save step can be elided:

  val side: SideInput[SparkeyReader] = sc
    .parallelize(Seq("a" -> "one", "b" -> "two"))
    .asSparkeySideInput

  SparkeyReader can be used like a lookup table in a side input operation:

  val main: SCollection[String] = sc.parallelize(Seq("a", "b", "c"))
  val side: SideInput[SparkeyReader] = sc
    .parallelize(Seq("a" -> "one", "b" -> "two"))
    .asSparkeySideInput
  
  main.withSideInputs(side)
    .map { (x, s) =>
      s(side).getOrElse(x, "unknown")
    }

  A SparkeyMap can store any types of keys and values, but can only be used as a SideInput:

  val main: SCollection[String] = sc.parallelize(Seq("a", "b", "c"))
  val side: SideInput[SparkeyMap[String, Int]] = sc
    .parallelize(Seq("a" -> 1, "b" -> 2, "c" -> 3))
    .asLargeMapSideInput()
  
  val objects: SCollection[MyObject] = main
    .withSideInputs(side)
    .map { (x, s) => s(side).get(x) }
    .toSCollection

  To read a static Sparkey collection and use it as a typed SideInput, use TypedSparkeyReader. TypedSparkeyReader can also accept a Caffeine cache to reduce IO and deserialization load:

  val main: SCollection[String] = sc.parallelize(Seq("a", "b", "c"))
  val cache: Cache[String, MyObject] = ...
  val side: SideInput[TypedSparkeyReader[MyObject]] = sc
    .typedSparkeySideInput("gs://<bucket>/<path>/<sparkey-prefix>", MyObject.decode, cache)
  
  val objects: SCollection[MyObject] = main
    .withSideInputs(side)
    .map { (x, s) => s(side).get(x) }
    .toSCollection

  Definition Classes

  extra

•  package instances

  Definition Classes

  sparkey

• InvalidNumShardsException
• LargeHashSCollectionFunctions
• PairLargeHashSCollectionFunctions
• SparkeyIO
• SparkeyPairSCollection
• SparkeySCollection
• SparkeyScioContext
• SparkeySetSCollection
• SparkeyUri
• SparkeyWritable

com.spotify.scio.extra.sparkey

SparkeyIO