[jira] [Updated] (S2GRAPH-252) Improve performance of S2GraphSource

["DOYUNG YOON (JIRA)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/S2GRAPH-252?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

DOYUNG YOON updated S2GRAPH-252:
--------------------------------
    Description: 
S2GraphSource is responsible to translate HBASE snapshot(*TableSnapshotInputFormat*) to graph element such as edge/vertex.

below code create *RDD[(ImmutableBytesWritable, Result)]* from *TableSnapshotInputFormat*

{noformat}
val rdd = ss.sparkContext.newAPIHadoopRDD(job.getConfiguration,
        classOf[TableSnapshotInputFormat],
        classOf[ImmutableBytesWritable],
        classOf[Result])
{noformat}

The problem comes after obtaining RDD. 

Current implementation use *RDD.mapPartitions* because S2Graph class is not serializable, mostly because it has Asynchbase client in it.

The problematic part is the following.

{noformat}
val elements = input.mapPartitions { iter =>
      val s2 = S2GraphHelper.getS2Graph(config)

      iter.flatMap { line =>
        reader.read(s2)(line)
      }
    }

    val kvs = elements.mapPartitions { iter =>
      val s2 = S2GraphHelper.getS2Graph(config)

      iter.map(writer.write(s2)(_))
    }
{noformat}

On each RDD partition, S2Graph instance connect meta storage, such as mysql, and use the local cache to avoid heavy read from meta storage.

Even though it works with a dataset with the small partition, the scalability of S2GraphSource limited by the number of partitions, which need to be increased when dealing with large data.

Possible improvement can be achieved by not depending on meta storage when it deserializes HBase's Result class into Edge/Vertex. 

We can simply achieve this by loading all necessary schemas from meta storage on spark driver, then broadcast these schemas and use them to deserialize instead of connecting meta storage on each partition.





  was:
S2GraphSource is responsible to translate HBASE snapshot(`TableSnapshotInputFormat`) to graph element such as edge/vertex.

below code create `RDD[(ImmutableBytesWritable, Result)]` from TableSnapshotInputFormat

{noformat}
val rdd = ss.sparkContext.newAPIHadoopRDD(job.getConfiguration,
        classOf[TableSnapshotInputFormat],
        classOf[ImmutableBytesWritable],
        classOf[Result])
{noformat}

The problem comes after obtaining RDD. 

Current implementation use `RDD.mapPartitions` because S2Graph class is not serializable, mostly because it has Asynchbase client in it.

The problematic part is the following.

{noformat}
val elements = input.mapPartitions { iter =>
      val s2 = S2GraphHelper.getS2Graph(config)

      iter.flatMap { line =>
        reader.read(s2)(line)
      }
    }

    val kvs = elements.mapPartitions { iter =>
      val s2 = S2GraphHelper.getS2Graph(config)

      iter.map(writer.write(s2)(_))
    }
{noformat}

On each RDD partition, S2Graph instance connect meta storage, such as mysql, and use the local cache to avoid heavy read from meta storage.

Even though it works with a dataset with the small partition, the scalability of S2GraphSource limited by the number of partitions, which need to be increased when dealing with large data.

Possible improvement can be achieved by not depending on meta storage when it deserializes HBase's Result class into Edge/Vertex. 

We can simply achieve this by loading all necessary schemas from meta storage on spark driver, then broadcast these schemas and use them to deserialize instead of connecting meta storage on each partition.






> Improve performance of S2GraphSource 
> -------------------------------------
>
>                 Key: S2GRAPH-252
>                 URL: https://issues.apache.org/jira/browse/S2GRAPH-252
>             Project: S2Graph
>          Issue Type: Improvement
>          Components: s2jobs
>            Reporter: DOYUNG YOON
>            Assignee: DOYUNG YOON
>            Priority: Major
>   Original Estimate: 336h
>  Remaining Estimate: 336h
>
> S2GraphSource is responsible to translate HBASE snapshot(*TableSnapshotInputFormat*) to graph element such as edge/vertex.
> below code create *RDD[(ImmutableBytesWritable, Result)]* from *TableSnapshotInputFormat*
> {noformat}
> val rdd = ss.sparkContext.newAPIHadoopRDD(job.getConfiguration,
>         classOf[TableSnapshotInputFormat],
>         classOf[ImmutableBytesWritable],
>         classOf[Result])
> {noformat}
> The problem comes after obtaining RDD. 
> Current implementation use *RDD.mapPartitions* because S2Graph class is not serializable, mostly because it has Asynchbase client in it.
> The problematic part is the following.
> {noformat}
> val elements = input.mapPartitions { iter =>
>       val s2 = S2GraphHelper.getS2Graph(config)
>       iter.flatMap { line =>
>         reader.read(s2)(line)
>       }
>     }
>     val kvs = elements.mapPartitions { iter =>
>       val s2 = S2GraphHelper.getS2Graph(config)
>       iter.map(writer.write(s2)(_))
>     }
> {noformat}
> On each RDD partition, S2Graph instance connect meta storage, such as mysql, and use the local cache to avoid heavy read from meta storage.
> Even though it works with a dataset with the small partition, the scalability of S2GraphSource limited by the number of partitions, which need to be increased when dealing with large data.
> Possible improvement can be achieved by not depending on meta storage when it deserializes HBase's Result class into Edge/Vertex. 
> We can simply achieve this by loading all necessary schemas from meta storage on spark driver, then broadcast these schemas and use them to deserialize instead of connecting meta storage on each partition.



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)