From 477d317a0ba52684be546079d93cb1ed5c00567a Mon Sep 17 00:00:00 2001
From: Seventeen17 <17aloha@gmail.com>
Date: Wed, 4 Jan 2023 22:54:25 +0800
Subject: [PATCH] feat: Add random_walk operator and node2vec model

---
 graphlearn/examples/tf/node2vec/dist_train.py | 158 +++++++++
 graphlearn/examples/tf/node2vec/node2vec.py   | 108 ++++++
 graphlearn/examples/tf/node2vec/train.py      | 167 ++++++++++
 graphlearn/python/c/py_export.cc              |   1 +
 graphlearn/python/config.py                   |   3 +
 graphlearn/python/data/decoder.py             |   8 +
 graphlearn/python/data/values.py              |   6 +
 graphlearn/python/gsl/dag_dataset.py          |  11 +-
 graphlearn/python/gsl/dag_node.py             |  41 ++-
 graphlearn/src/common/base/config.cc          |   2 +
 .../core/operator/random_walk/random_walk.cc  | 273 +++++++++++++++
 .../src/core/partition/hash_partitioner.h     |  44 ++-
 graphlearn/src/core/partition/stitcher.h      |  27 +-
 graphlearn/src/include/client.h               |   2 +
 graphlearn/src/include/config.h               |   2 +
 graphlearn/src/include/constants.h            |   1 +
 graphlearn/src/include/random_walk_request.h  | 105 ++++++
 graphlearn/src/service/client.cc              |   1 +
 graphlearn/src/service/constants.cc           |   1 +
 .../service/request/random_walk_request.cc    | 310 ++++++++++++++++++
 20 files changed, 1252 insertions(+), 19 deletions(-)
 create mode 100644 graphlearn/examples/tf/node2vec/dist_train.py
 create mode 100644 graphlearn/examples/tf/node2vec/node2vec.py
 create mode 100644 graphlearn/examples/tf/node2vec/train.py
 create mode 100644 graphlearn/src/core/operator/random_walk/random_walk.cc
 create mode 100644 graphlearn/src/include/random_walk_request.h
 create mode 100644 graphlearn/src/service/request/random_walk_request.cc

diff --git a/graphlearn/examples/tf/node2vec/dist_train.py b/graphlearn/examples/tf/node2vec/dist_train.py
new file mode 100644
index 00000000..fda9f2a6
--- /dev/null
+++ b/graphlearn/examples/tf/node2vec/dist_train.py
@@ -0,0 +1,158 @@
+# Copyright 2022 Alibaba Group Holding Limited. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# =============================================================================
+from __future__ import print_function
+
+import os
+import sys
+
+import numpy as np
+import graphlearn as gl
+import tensorflow as tf
+import graphlearn.python.nn.tf as tfg
+
+sys.path.append("..")
+from node2vec import Node2Vec
+from trainer import DistTrainer
+
+# Settings
+flags = tf.app.flags
+FLAGS = flags.FLAGS
+flags.DEFINE_integer('task_index', None, 'Task index')
+flags.DEFINE_integer('task_count', None, 'Task count')
+flags.DEFINE_string('job_name', None, 'worker or ps')
+flags.DEFINE_string('ps_hosts', '', 'ps hosts')
+flags.DEFINE_string('worker_hosts', '', 'worker hosts')
+flags.DEFINE_string('tables', '', 'input odps tables names of node and edge.')
+flags.DEFINE_string('outputs', '', 'ouput odps tables name of node embeddings')
+flags.DEFINE_string('buckets', '', 'oss buckets for ckpt')
+flags.DEFINE_string('ckpt_dir', '', 'oss buckets for ckpt')
+# user-defined params
+flags.DEFINE_integer('epochs', 10, 'training epochs')
+flags.DEFINE_integer('node_count', 10312, 'count of nodes')
+flags.DEFINE_bool('need_hash', True, 'is node id continues or not,'
+                  'if continues, need_hash is False and id should be in [0, node_count).')
+flags.DEFINE_integer('batch_size', 128, 'minibatch size')
+flags.DEFINE_float('learning_rate', 0.01, 'learning rate')
+flags.DEFINE_integer('output_dim', 128, 'final output embedding dim')
+flags.DEFINE_integer('neg_num', 10, 'negative sampling num')
+flags.DEFINE_integer('walk_len', 10, 'random walk length')
+flags.DEFINE_integer('window_size', 2, 'window size for pairing in random walks')
+flags.DEFINE_bool('profiling', True, 'record timeline or not')
+flags.DEFINE_float('param_return', 0.25, 'param return')
+flags.DEFINE_float('param_inout', 0.25, 'param inout')
+flags.DEFINE_integer('max_neighbor_count', 100000, 'max nerighbor count')
+
+# global settings.
+gl.set_retry_times(30)
+gl.set_timeout(300)
+gl.set_tape_capacity(10)
+gl.set_default_full_nbr_num(FLAGS.max_neighbor_count)
+tfg.conf.emb_max_partitions = len(FLAGS.ps_hosts.split(',')) # embedding varible partition num.
+
+def load_graph():
+  node_table, edge_table = FLAGS.tables.split(',')[0:2]
+
+  g = gl.Graph() \
+    .node(node_table, node_type='i', decoder=gl.Decoder()) \
+    .edge(edge_table, edge_type=('i', 'i', 'i-i'),
+          decoder=gl.Decoder(weighted=True), directed=False)
+  return g
+
+def random_walk(graph):
+  """ random walk along with i-i edges, and return walks and their negative neighbors
+  as tensorflow tensors.
+  Shape of returned walks is [B, W], and shape of their negative neighbors
+  is [B, W, N]. While B=batch size, W=walk length, N=negtiave nums.
+  """
+  # B: batch_size, W: walk_len, N: neg_num
+  p = FLAGS.param_return
+  q = FLAGS.param_inout
+  src = graph.V('i').batch(FLAGS.batch_size).shuffle(traverse=True).alias('src')
+  walks = src.random_walk('i-i', FLAGS.walk_len - 1, p=p, q=q).alias('walks')
+
+  neg = src.outNeg('i-i').sample(FLAGS.neg_num).by('random').alias('neg')
+  walks_neg = walks.outNeg('i-i').sample(FLAGS.neg_num).by('random').alias('walks_neg')
+  query = src.values()
+
+  dataset = tfg.Dataset(query, 10)
+  data_dict = dataset.get_data_dict()
+
+  src = tf.expand_dims(data_dict['src'].ids, 1) # [B, 1]
+  walks = tf.reshape(data_dict['walks'].ids, (-1, FLAGS.walk_len - 1)) # [B, W - 1]
+  walks = tf.concat([src, walks], axis=1) # [B, W]
+
+  neg = tf.expand_dims(tf.reshape(data_dict['neg'].ids,
+                                  (-1, FLAGS.neg_num)), axis=1) # [B, 1, N]
+  neg_dst_walks = tf.reshape(data_dict['walks_neg'].ids,
+                             (-1, FLAGS.walk_len - 1, FLAGS.neg_num)) # # [B, W - 1, N]
+  negs = tf.concat([neg, neg_dst_walks], axis=1) # [B, W, N]
+  return dataset.iterator, walks, negs # [B, W], [B, W, N]
+
+def train(graph, model):
+  tfg.conf.training = True
+  iterator, walks, negs = random_walk(graph)
+
+  embs = model.forward(walks, negs)
+  loss = model.loss(*embs)
+  return iterator, loss
+
+def save_node_embedding(graph, model):
+  tfg.conf.training = False
+  query_save = graph.V('i').batch(FLAGS.batch_size).alias('i').values()
+  dataset = tfg.Dataset(query_save, 10)
+  data_dict = dataset.get_data_dict()
+  emb = model.forward(data_dict['i'].ids)
+  return dataset.iterator, data_dict['i'].ids, emb
+
+def main():
+  worker_count = len(FLAGS.worker_hosts.split(','))
+  ps_count = len(FLAGS.ps_hosts.split(','))
+  graph_cluster = {'client_count': worker_count, 'server_count': ps_count}
+  g = load_graph()
+  if FLAGS.job_name == 'worker':
+    g.init(cluster=graph_cluster, job_name='client', task_index=FLAGS.task_index)
+  else:
+    g.init(cluster=graph_cluster, job_name='server', task_index=FLAGS.task_index)
+
+  ps_hosts = FLAGS.ps_hosts.split(',')
+  worker_hosts = FLAGS.worker_hosts.split(',')
+  tf_cluster = tf.train.ClusterSpec({'ps': ps_hosts, 'worker': worker_hosts})
+  trainer = DistTrainer(tf_cluster,
+                        FLAGS.job_name,
+                        FLAGS.task_index,
+                        worker_count,
+                        ckpt_dir=FLAGS.buckets,
+                        profiling=FLAGS.profiling)
+  if FLAGS.job_name == 'worker':
+    with trainer.context(): # model must under trainer.context.
+      model = Node2Vec(FLAGS.walk_len,
+                       FLAGS.window_size,
+                       FLAGS.window_size,
+                       FLAGS.node_count,
+                       FLAGS.need_hash,
+                       FLAGS.output_dim,
+                       neg_num=FLAGS.neg_num)
+      train_iterator, loss = train(g, model)
+      save_iterator, ids, emb = save_node_embedding(g, model)
+    # training
+    trainer.train(train_iterator, loss, FLAGS.learning_rate, epochs=FLAGS.epochs)
+    # saving node embedding with your own writer.
+  else:
+    trainer.join()
+  g.close()
+  print('Finished!')
+
+if __name__ == "__main__":
+  main()
diff --git a/graphlearn/examples/tf/node2vec/node2vec.py b/graphlearn/examples/tf/node2vec/node2vec.py
new file mode 100644
index 00000000..8862dd1b
--- /dev/null
+++ b/graphlearn/examples/tf/node2vec/node2vec.py
@@ -0,0 +1,108 @@
+# Copyright 2022 Alibaba Group Holding Limited. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# =============================================================================
+from __future__ import absolute_import
+from __future__ import division
+from __future__ import print_function
+
+import tensorflow as tf
+import graphlearn.python.nn.tf as tfg
+
+class Node2Vec(tfg.Module):
+  def __init__(self,
+               walk_len,
+               l,
+               r,
+               bucket_size,
+               need_hash,
+               dim,
+               **kwargs):
+    """Node2Vec model.
+    Args:
+      walk_len: integer, random walk steps
+      l: left window size of random walk pair generated.
+      r: right window size of random walk pair generated.
+      bucket_size: embedding bucket size
+      dim: embedding dimension
+    """
+    self.walk_len = walk_len
+    self.l = l
+    self.r = r
+    self.dim = dim
+
+    self._target_encoder = tfg.EmbeddingColumn('src_emb', bucket_size, dim, need_hash)
+    self._context_encoder = tfg.EmbeddingColumn('dst_emb', bucket_size, dim, need_hash)
+    self._neg_num = kwargs.get('neg_num', 1)
+
+  def _pair_indices(self):
+    """ Generate indices of pairs from random walks, with left window
+    size and right window size.
+    """
+    indices_src = []
+    indices_dst = []
+    for i in range(self.walk_len):
+      for j in range(max(i - self.l, 0), i):
+        indices_src.append(i)
+        indices_dst.append(j)
+      for j in range(i + 1, min(i + self.r + 1, self.walk_len)):
+        indices_src.append(i)
+        indices_dst.append(j)
+    return indices_src, indices_dst
+
+  def forward(self, x, neg=None):
+    """ Encode tensor ids, if neg is None, encode x with target_encoder,
+    else, encode x with both target_encoder and context_encoder, and encode
+    neg with context_encoder, and return a tuple.
+    """
+    if neg is None:
+      return self._target_encoder(tf.reshape(x, [-1]))
+    else:
+      return self._target_encoder(tf.reshape(x, [-1])), \
+             self._context_encoder(tf.reshape(x, [-1])), \
+             self._context_encoder(tf.reshape(neg, [-1]))
+
+  def loss(self, walks_emb_src, walks_emb_dst, neg_emb):
+    """ First gather embeddings from walks_emb_src and walks_emb_dst between
+    window (left_window_size, right_window_size) as pairs, and then calculate
+    pointwise sigmoid cross entropy loss.
+
+    Args:
+      walks_emb_src: embedding of walks encoded by target_encoder, shape=[-1, D]
+      walks_emb_dst: embedding of walks encoded by context_encoder, shape=[-1, D]
+      neg_emb: embedding of negative samples encoded by context_encoder, shape=[-1, D]
+    """
+    path_src = tf.reshape(walks_emb_src, shape=[-1, self.walk_len, self.dim])
+    path_dst = tf.reshape(walks_emb_dst, shape=[-1, self.walk_len, self.dim])
+    neg_dst = tf.reshape(neg_emb, shape=[-1, self._neg_num, self.dim])
+    src_indices, dst_indices = self._pair_indices()
+
+    src = tf.reshape(tf.gather(path_src, src_indices, axis=1),
+                     shape=[-1, self.dim])
+    dst = tf.reshape(tf.gather(path_dst, dst_indices, axis=1),
+                     shape=[-1, self.dim])
+
+    pos_logit = tf.reduce_sum(
+      tf.multiply(src, dst, name='rank_dot_prod'), axis=-1)
+    true_xent = tf.nn.sigmoid_cross_entropy_with_logits(
+      labels=tf.ones_like(pos_logit), logits=pos_logit)
+    neg_logit = tf.reduce_sum(
+      tf.multiply(
+        tf.tile(tf.expand_dims(walks_emb_src, axis=1), [1, self._neg_num, 1]),
+        neg_dst, name='rank_dot_prod'),
+      axis=-1)
+    neg_xent = tf.nn.sigmoid_cross_entropy_with_logits(
+      labels=tf.zeros_like(neg_logit), logits=neg_logit)
+    lo = tf.reduce_mean(true_xent) + tf.reduce_mean(neg_xent)
+    return lo
+
diff --git a/graphlearn/examples/tf/node2vec/train.py b/graphlearn/examples/tf/node2vec/train.py
new file mode 100644
index 00000000..96459e0b
--- /dev/null
+++ b/graphlearn/examples/tf/node2vec/train.py
@@ -0,0 +1,167 @@
+# Copyright 2022 Alibaba Group Holding Limited. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# =============================================================================
+from __future__ import print_function
+
+import os
+import sys
+
+import graphlearn as gl
+import graphlearn.python.nn.tf as tfg
+import numpy as np
+import tensorflow as tf
+
+from node2vec import Node2Vec
+sys.path.append(os.path.dirname(sys.path[0]))
+
+from tensorflow.python.client import timeline
+run_options = tf.RunOptions(trace_level=tf.RunOptions.FULL_TRACE)
+run_metadata = tf.RunMetadata()
+
+# Settings
+flags = tf.app.flags
+FLAGS = flags.FLAGS
+flags.DEFINE_string('tables', '../../data/blogcatelog/node_table,../../data/blogcatelog/edge_table',
+                    'input odps tables names of node and edge.')
+flags.DEFINE_string('ckpt_dir', './', 'dir for ckpt')
+# user-defined params
+flags.DEFINE_integer('epochs', 40, 'training epochs')
+flags.DEFINE_integer('node_count', 10312, 'count of nodes')
+flags.DEFINE_bool('need_hash', False, 'is node id continues or not,'
+                  'if continues, need_hash is False and id should be in [0, node_count).')
+flags.DEFINE_integer('batch_size', 256, 'minibatch size')
+flags.DEFINE_float('learning_rate', 0.01, 'learning rate')
+flags.DEFINE_integer('output_dim', 128, 'final output embedding dim')
+flags.DEFINE_integer('neg_num', 10, 'negative sampling num')
+flags.DEFINE_integer('walk_len', 20, 'random walk length')
+flags.DEFINE_integer('window_size', 2, 'window size')
+flags.DEFINE_float('param_return', 0.25, 'param return')
+flags.DEFINE_float('param_inout', 0.25, 'param inout')
+
+def load_graph():
+  node_table, edge_table = FLAGS.tables.split(',')[0:2]
+
+  g = gl.Graph() \
+    .node(node_table, node_type='i',
+          decoder=gl.Decoder(weighted=True)) \
+    .edge(edge_table, edge_type=('i', 'i', 'i-i'),
+          decoder=gl.Decoder(weighted=True), directed=False)
+  return g
+
+def random_walk(graph):
+  """ random walk along with i-i edges, and return walks and their negative neighbors
+  as tensorflow tensors.
+  Shape of returned walks is [B, W], and shape of their negative neighbors
+  is [B, W, N]. While B=batch size, W=walk length, N=negtiave nums.
+  """
+  # B: batch_size, W: walk_len, N: neg_num
+  p = FLAGS.param_return
+  q = FLAGS.param_inout
+  src = graph.V('i').batch(FLAGS.batch_size).shuffle(traverse=True).alias('src')
+  walks = src.random_walk('i-i', FLAGS.walk_len - 1, p=p, q=q).alias('walks')
+
+  neg = src.outNeg('i-i').sample(FLAGS.neg_num).by('random').alias('neg')
+  walks_neg = walks.outNeg('i-i').sample(FLAGS.neg_num).by('random').alias('walks_neg')
+  query = src.values()
+
+  dataset = tfg.Dataset(query, 10)
+  data_dict = dataset.get_data_dict()
+
+  src = tf.expand_dims(data_dict['src'].ids, 1) # [B, 1]
+  walks = tf.reshape(data_dict['walks'].ids, (-1, FLAGS.walk_len - 1)) # [B, W - 1]
+  walks = tf.concat([src, walks], axis=1) # [B, W]
+
+  neg = tf.expand_dims(tf.reshape(data_dict['neg'].ids,
+                                  (-1, FLAGS.neg_num)), axis=1) # [B, 1, N]
+  neg_dst_walks = tf.reshape(data_dict['walks_neg'].ids,
+                             (-1, FLAGS.walk_len - 1, FLAGS.neg_num)) # # [B, W - 1, N]
+  negs = tf.concat([neg, neg_dst_walks], axis=1) # [B, W, N]
+  return dataset.iterator, walks, negs # [B, W], [B, W, N]
+
+def train(graph, model):
+  tfg.conf.training = True
+  iterator, walks, negs = random_walk(graph)
+
+  embs = model.forward(walks, negs)
+  loss = model.loss(*embs)
+  return iterator, loss
+
+def save_node_embedding(graph, model):
+  tfg.conf.training = False
+  query_save = graph.V('i').batch(FLAGS.batch_size).alias('i').values()
+  dataset = tfg.Dataset(query_save, 10)
+  data_dict = dataset.get_data_dict()
+  emb = model.forward(data_dict['i'].ids)
+  return dataset.iterator, data_dict['i'].ids, emb
+
+def main():
+  g = load_graph()
+  g.init()
+
+  model = Node2Vec(FLAGS.walk_len,
+                   FLAGS.window_size,
+                   FLAGS.window_size,
+                   FLAGS.node_count,
+                   FLAGS.need_hash,
+                   FLAGS.output_dim,
+                   neg_num=FLAGS.neg_num)
+  train_iterator, loss = train(g, model)
+  optimizer=tf.train.AdamOptimizer(learning_rate=FLAGS.learning_rate)
+  train_op = optimizer.minimize(loss)
+  train_ops = [loss, train_op]
+  save_iterator, ids, emb = save_node_embedding(g, model)
+  with tf.Session() as sess:
+    sess.run(tf.local_variables_initializer())
+    sess.run(tf.global_variables_initializer())
+    sess.run(train_iterator.initializer)
+    step = 0
+    print("Start Training...")
+    for i in range(FLAGS.epochs):
+      try:
+        while True:
+          if step < 20 and step > 10:
+            ret = sess.run(train_ops,
+                           options=run_options,
+                           run_metadata=run_metadata)
+            tl = timeline.Timeline(run_metadata.step_stats)
+            content = tl.generate_chrome_trace_format()
+            file_name = 'timeline_' + str(step) + '.json'
+            save_path = os.path.join(FLAGS.ckpt_dir, file_name)
+            writeGFile = tf.gfile.GFile(save_path, mode='w')
+            writeGFile.write(content)
+            writeGFile.flush()
+            writeGFile.close()
+          else:
+            ret = sess.run(train_ops)
+          print("Epoch {}, Iter {}, Loss {:.5f}".format(i, step, ret[0]))
+          step += 1
+      except tf.errors.OutOfRangeError:
+        sess.run(train_iterator.initializer) # reinitialize dataset.
+
+    sess.run(save_iterator.initializer)
+    emb_set = []
+    while True:
+      try:
+        rets = sess.run([ids, emb])
+        emb_set.append(np.concatenate([np.reshape(rets[0], [-1, 1]), rets[1]],
+                                      axis=1))
+      except tf.errors.OutOfRangeError:
+        print("Save node embedding finished.")
+        break
+    emb_set = np.concatenate(emb_set, axis=0)
+    np.save('id_emb.npy', emb_set)
+  g.close()
+
+if __name__ == "__main__":
+  main()
diff --git a/graphlearn/python/c/py_export.cc b/graphlearn/python/c/py_export.cc
index b8d84460..c2949a8e 100644
--- a/graphlearn/python/c/py_export.cc
+++ b/graphlearn/python/c/py_export.cc
@@ -36,6 +36,7 @@ PYBIND11_MODULE(pywrap_graphlearn, m) {
   m.doc() = "Python interface for graph-learn.";
   // setters
   m.def("set_default_neighbor_id", &SetGlobalFlagDefaultNeighborId);
+  m.def("set_default_full_nbr_num", &SetGlobalFlagDefaultFullNbrNum);
   m.def("set_tracker_mode", &SetGlobalFlagTrackerMode);
   m.def("set_padding_mode", &SetGlobalFlagPaddingMode);
   m.def("set_storage_mode", &SetGlobalFlagStorageMode);
diff --git a/graphlearn/python/config.py b/graphlearn/python/config.py
index 002d5f5d..c3c9a2fa 100644
--- a/graphlearn/python/config.py
+++ b/graphlearn/python/config.py
@@ -23,6 +23,9 @@
 def set_default_neighbor_id(nbr_id):
   pywrap.set_default_neighbor_id(nbr_id)
 
+def set_default_full_nbr_num(num):
+  pywrap.set_default_full_nbr_num(num)
+
 def set_tracker_mode(mode):
   pywrap.set_tracker_mode(mode)
 
diff --git a/graphlearn/python/data/decoder.py b/graphlearn/python/data/decoder.py
index c2bd2a53..7e6dd103 100644
--- a/graphlearn/python/data/decoder.py
+++ b/graphlearn/python/data/decoder.py
@@ -97,6 +97,10 @@ def __init__(self,
 
     self._attributed = self._parse_attributes()
 
+    self._has_property = False
+    if self._weighted or self._labeled or self._timestamped or self._attributed:
+      self._has_property = True
+
   def _parse_attributes(self):
     if not self._attr_types:
       return False
@@ -188,6 +192,10 @@ def data_format(self):
     return int(self._weighted * 2 + \
                self._labeled * 4 + self._attributed * 8)
 
+  @property
+  def has_property(self):
+    return self._has_property
+
   @property
   def int_attr_num(self):
     return self._int_attr_num
diff --git a/graphlearn/python/data/values.py b/graphlearn/python/data/values.py
index eb2e23d3..3bc91a5d 100644
--- a/graphlearn/python/data/values.py
+++ b/graphlearn/python/data/values.py
@@ -154,6 +154,12 @@ def _init(self):
     if self._inited:
       return
 
+    try:
+      if not self._get_decoder().has_property:
+        return
+    except AttributeError:
+      print('Get Decoder for {} failed.'.format(self._type))
+
     values = self._lookup()
     self.int_attrs = values.int_attrs
     self.float_attrs = values.float_attrs
diff --git a/graphlearn/python/gsl/dag_dataset.py b/graphlearn/python/gsl/dag_dataset.py
index c7a36834..1bb5ffe4 100644
--- a/graphlearn/python/gsl/dag_dataset.py
+++ b/graphlearn/python/gsl/dag_dataset.py
@@ -122,11 +122,12 @@ def __getitem__(self, alias):
 
     # Add the attributes for the Nodes/Edges.
     lookup_node = node.get_lookup_node()
-    res.int_attrs = pywrap.get_dag_value(self._res, lookup_node.nid, "ia")
-    res.float_attrs = pywrap.get_dag_value(self._res, lookup_node.nid, "fa")
-    res.string_attrs = pywrap.get_dag_value(self._res, lookup_node.nid, "sa")
-    res.weights = pywrap.get_dag_value(self._res, lookup_node.nid, "wei")
-    res.labels = pywrap.get_dag_value(self._res, lookup_node.nid, "lb")
+    if lookup_node is not None:
+      res.int_attrs = pywrap.get_dag_value(self._res, lookup_node.nid, "ia")
+      res.float_attrs = pywrap.get_dag_value(self._res, lookup_node.nid, "fa")
+      res.string_attrs = pywrap.get_dag_value(self._res, lookup_node.nid, "sa")
+      res.weights = pywrap.get_dag_value(self._res, lookup_node.nid, "wei")
+      res.labels = pywrap.get_dag_value(self._res, lookup_node.nid, "lb")
 
     # Add degrees for the Nodes.
     for dg_node in node.get_degree_nodes():
diff --git a/graphlearn/python/gsl/dag_node.py b/graphlearn/python/gsl/dag_node.py
index d1d5740a..c9138acb 100644
--- a/graphlearn/python/gsl/dag_node.py
+++ b/graphlearn/python/gsl/dag_node.py
@@ -44,6 +44,7 @@ def __init__(self, dag, op_name="", params={}):
     self._path = None  # Upstream edge type
     self._type = None  # Edge type or node type of current DagNode
     self._strategy = "random"
+    self._remove_property = False
 
     # Add by GSL.
     # All the DagEdges expect link to sink node are mapped to DagEdgeDef.
@@ -200,7 +201,7 @@ def by(self, strategy):
       assert strategy in ["random", "in_degree", "conditional", "node_weight"]
     elif self._op_name == "Sampler":
       assert strategy in \
-        ["random", "topk", "in_degree", "edge_weight", "full"]
+        ["random", "topk", "in_degree", "edge_weight", "full", "random_without_replacement"]
     else:
       raise ValueError("`by(strategy)` can only be used after`sample(count)`")
 
@@ -210,7 +211,7 @@ def by(self, strategy):
     return self
 
   def filter(self, target):
-    """Filter the samples that are not equal to target ids.
+    """Filter the samples that are equal to target ids.
 
     Args:
       target (string): Alias of upstream TraverseVertexDagNode.
@@ -294,6 +295,10 @@ def each(self, func):
     func(self)
     return self
 
+  def remove_property(self):
+    self._remove_property = True
+    return self
+
   def values(self, func=lambda x: x):
     self._dag.set_ready(func)
     return self._dag
@@ -308,7 +313,10 @@ def _set_alias(self, alias=None, temp=False):
     self._alias = alias
     self._dag.add_node(alias, self, temp=temp)
 
-    self._lookup_node = self._lookup()
+    if self._remove_property:
+      self._lookup_node = None
+    else:
+      self._lookup_node = self._lookup()
     self._link_to_sink()
 
   def _get_shape_and_degrees(self, dag_values):
@@ -355,22 +363,28 @@ def _new_edge_node(self, op_name, edge_type, in_edge):
 
     next_node.set_path(edge_type, pywrap.NodeFrom.NODE)
     next_node.set_output_field(pywrap.kEdgeIds)
+
     return next_node
 
   def _new_vertex_node(self, op_name, edge_type, in_edge,
                        node_from=pywrap.NodeFrom.EDGE_DST):
     assert edge_type is not None and isinstance(edge_type, str)
 
-    self._add_out_edge(in_edge)
+    if not isinstance(in_edge, list):
+      in_edge = [in_edge]
+    for e in in_edge:
+      self._add_out_edge(e)
     shape = self._shape
 
     next_node = TraverseVertexDagNode(self._dag, op_name=op_name)
     next_node._shape = shape
     next_node._add_param(pywrap.kEdgeType, edge_type)
-    next_node._add_in_edge(in_edge)
+    for e in in_edge:
+      next_node._add_in_edge(e)
 
     next_node.set_path(edge_type, node_from)
     next_node.set_output_field(pywrap.kNodeIds)
+
     return next_node
 
   def set_ready(self, node_id):
@@ -495,6 +509,22 @@ def Neg(self, node_type):
     self._neg_downstreams.append(next_node)
     return next_node
 
+  def random_walk(self, edge_type, walk_len=1, p=1.0, q=1.0):
+    walk_len = int(walk_len)
+    assert(walk_len > 0)
+    self._set_alias()
+    in_edge_src = self._new_edge(dst_input=pywrap.kSrcIds)
+    in_edge_parent = self._new_edge(dst_input=pywrap.kNodeIds)
+    next_node = self._new_vertex_node("RandomWalk", edge_type,
+                                      [in_edge_src, in_edge_parent])
+    next_node._add_param(pywrap.kSideInfo, [float(p), float(q)])
+    next_node._add_param(pywrap.kDistances, int(walk_len))
+    next_node._add_param(pywrap.kPartitionKey, pywrap.kSrcIds)
+    next_node._shape = (np.prod(self._shape), walk_len)
+    next_node._remove_property = True
+    self._pos_downstreams.append(next_node)
+    return next_node
+
   def _lookup(self):
     # Override
     # Generate an edge from traverse node to it's lookup node.
@@ -611,7 +641,6 @@ def feed_values(self, dag_values):
     return self._graph.get_edges(
         self._type, src_ids, dst_ids, edge_ids, degrees, shape)
 
-
 class LookupDagNode(DagNode):
   def __init__(self, op_name="", upstream=None, in_edges=[], params={}):
     super(LookupDagNode, self).__init__(upstream._dag, op_name, params)
diff --git a/graphlearn/src/common/base/config.cc b/graphlearn/src/common/base/config.cc
index 303cf62f..919e5c9f 100644
--- a/graphlearn/src/common/base/config.cc
+++ b/graphlearn/src/common/base/config.cc
@@ -96,6 +96,7 @@ DEFINE_INT32_GLOBAL_FLAG(TrackerMode, 1)  // 0: Rpc, 1: FileSystem
 DEFINE_INT64_GLOBAL_FLAG(AverageNodeCount, 10000)
 DEFINE_INT64_GLOBAL_FLAG(AverageEdgeCount, 10000)
 DEFINE_INT64_GLOBAL_FLAG(DefaultNeighborId, 0)
+DEFINE_INT32_GLOBAL_FLAG(DefaultFullNbrNum, 5000)
 DEFINE_INT64_GLOBAL_FLAG(DefaultIntAttribute, 0)
 DEFINE_FLOAT_GLOBAL_FLAG(DefaultFloatAttribute, 0.0)
 DEFINE_STRING_GLOBAL_FLAG(DefaultStringAttribute, "")
@@ -135,6 +136,7 @@ DEFINE_SET_INT32_GLOBAL_FLAG(TrackerMode)
 DEFINE_SET_INT64_GLOBAL_FLAG(AverageNodeCount)
 DEFINE_SET_INT64_GLOBAL_FLAG(AverageEdgeCount)
 DEFINE_SET_INT64_GLOBAL_FLAG(DefaultNeighborId)
+DEFINE_SET_INT32_GLOBAL_FLAG(DefaultFullNbrNum)
 DEFINE_SET_INT64_GLOBAL_FLAG(DefaultIntAttribute)
 DEFINE_SET_FLOAT_GLOBAL_FLAG(DefaultFloatAttribute)
 DEFINE_SET_STRING_GLOBAL_FLAG(DefaultStringAttribute)
diff --git a/graphlearn/src/core/operator/random_walk/random_walk.cc b/graphlearn/src/core/operator/random_walk/random_walk.cc
new file mode 100644
index 00000000..d66207cd
--- /dev/null
+++ b/graphlearn/src/core/operator/random_walk/random_walk.cc
@@ -0,0 +1,273 @@
+/* Copyright 2022 Alibaba Group Holding Limited. All Rights Reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+==============================================================================*/
+
+#include <cmath>
+#include <random>
+#include "include/random_walk_request.h"
+#include "core/operator/op_registry.h"
+#include "core/operator/operator.h"
+#include "core/operator/sampler/alias_method.h"
+#include "core/runner/op_runner.h"
+#include "include/client.h"
+#include "include/config.h"
+#include "include/constants.h"
+
+
+namespace graphlearn {
+namespace op {
+
+class RandomWalk : public RemoteOperator {
+public:
+  virtual ~RandomWalk() {}
+
+  Status Process(const OpRequest* req,
+                 OpResponse* res) override {
+    const RandomWalkRequest* request =
+      static_cast<const RandomWalkRequest*>(req);
+    RandomWalkResponse* response =
+      static_cast<RandomWalkResponse*>(res);
+
+    Status s;
+    auto walk_len = request->WalkLen();
+    auto batch_size = request->BatchSize();
+
+    size_t total_size = batch_size * walk_len;
+    response->SetBatchSize(batch_size);
+    response->InitWalks(total_size);
+
+    std::vector<int64_t> ret_walks;
+    ret_walks.reserve(total_size);
+
+    std::unique_ptr<OpRunner> runner = GetOpRunner(Env::Default(), this);
+
+    if (request->IsDeepWalk()) {
+      // Walk the 1st steps locally.
+      this->DeepWalk(request, &ret_walks);
+      if (walk_len <= 1) {
+        response->AppendWalks(ret_walks.data(), total_size);
+        return s;
+      }
+
+      // Walk the 2-N steps remote.
+      for (int32_t step = 1; step < walk_len; ++step) {
+        RandomWalkRequest sub_req(request->Type(), request->P(), request->Q());
+        RandomWalkResponse sub_res;
+
+        std::vector<int64_t> sub_walks;
+        sub_walks.reserve(batch_size);
+        auto begin = ret_walks.begin() + batch_size * (step - 1);
+        std::copy(begin, begin + batch_size, std::back_inserter(sub_walks));
+        sub_req.Set(sub_walks.data(), sub_walks.size());
+
+        s = runner->Run(&sub_req, &sub_res);
+        if (!s.ok()) return s;
+        std::copy(sub_res.GetWalks(), sub_res.GetWalks() + batch_size, std::back_inserter(ret_walks));
+      }
+    } else {
+      std::vector<int64_t> ret_nbrs;
+      std::vector<int32_t> ret_degrees;
+      int32_t total_neighbor_count = 0; // Total neighbor count for the batch src_ids.
+      ret_nbrs.reserve(GLOBAL_FLAG(DefaultFullNbrNum) * batch_size);
+      ret_degrees.reserve(batch_size);
+
+      this->WeightedRandomWalk(request, &ret_walks, &ret_nbrs, &ret_degrees);
+
+      for (int32_t i = 0; i < batch_size; ++i) {
+        total_neighbor_count += ret_degrees[i];
+      }
+
+      if (walk_len <= 1) {
+        response->AppendWalks(ret_walks.data(), total_size);
+        response->SetSparseFlag();
+        response->InitNeighborIds(total_neighbor_count);
+        response->InitDegrees(batch_size);
+        response->AppendNeighborIds(ret_nbrs.data(), total_neighbor_count);
+        response->AppendDegrees(ret_degrees.data(), batch_size);
+        return s;
+      }
+
+      for (int32_t step = 1; step < walk_len; ++step) {
+        RandomWalkRequest sub_req(request->Type(), request->P(), request->Q());
+        RandomWalkResponse sub_res;
+
+        std::vector<int64_t> sub_walks;
+        sub_walks.reserve(batch_size);
+        std::copy(ret_walks.begin() + batch_size * (step - 1), ret_walks.begin() + batch_size * step,
+                  std::back_inserter(sub_walks));
+        std::vector<int64_t> sub_src;
+        sub_src.reserve(batch_size);
+        auto begin = ret_walks.begin() + batch_size * std::max(step - 2, 0);
+        std::copy(begin, begin + batch_size, std::back_inserter(sub_src));
+        sub_req.Set(sub_walks.data(),
+                    sub_src.data(),
+                    batch_size,
+                    ret_nbrs.data(),
+                    ret_degrees.data(),
+                    total_neighbor_count);
+        s = runner->Run(&sub_req, &sub_res);
+        if (!s.ok()) return s;
+
+        auto degrees = sub_res.GetDegrees();
+        total_neighbor_count = 0;
+        for (int32_t i = 0; i < batch_size; ++i) {
+          total_neighbor_count += *(degrees + i);
+        }
+        std::copy(sub_res.GetWalks(), sub_res.GetWalks() + batch_size, std::back_inserter(ret_walks));
+        ret_nbrs.clear();
+        ret_degrees.clear();
+        std::copy(sub_res.GetNeighborIds(), sub_res.GetNeighborIds() + total_neighbor_count,  std::back_inserter(ret_nbrs));
+        std::copy(degrees, degrees + batch_size, std::back_inserter(ret_degrees));
+      }
+
+    }
+
+    int64_t ret[total_size];
+    for (size_t row = 0; row < walk_len; ++row) {
+      for (size_t col = 0; col < batch_size; ++col) {
+        ret[col * walk_len + row] = ret_walks[row * batch_size + col];
+      }
+    }
+
+    response->AppendWalks(ret, total_size);
+    return s;
+  }
+
+  Status Call(int32_t remote_id,
+              const OpRequest* req,
+              OpResponse* res) override {
+    const RandomWalkRequest* request =
+      static_cast<const RandomWalkRequest*>(req);
+    RandomWalkResponse* response =
+      static_cast<RandomWalkResponse*>(res);
+    std::unique_ptr<Client> client(NewRpcClient(remote_id));
+    return client->RandomWalk(request, response);
+  }
+
+private:
+void DeepWalk(const RandomWalkRequest* req,
+             std::vector<int64_t>* ret_walks) {
+  int32_t batch_size = req->BatchSize();
+  auto src_ids = req->GetSrcIds();
+
+  const std::string& edge_type = req->Type();
+  Graph* graph = graph_store_->GetGraph(edge_type);
+  auto storage = graph->GetLocalStorage();
+
+  for (int32_t i = 0; i < batch_size; ++i) {
+      auto src_id = src_ids[i];
+      auto neighbor_ids = storage->GetNeighbors(src_id);
+
+      auto nbr_count = neighbor_ids.Size();
+      if (nbr_count == 0) {
+        ret_walks->push_back(GLOBAL_FLAG(DefaultNeighborId));
+      } else {
+        thread_local static std::random_device rd;
+        thread_local static std::mt19937 engine(rd());
+        std::uniform_int_distribution<> dist(0, nbr_count - 1);
+        int32_t indice = dist(engine);
+        ret_walks->push_back(neighbor_ids[indice]);
+      }
+    }
+  }
+
+  void WeightedRandomWalk(const RandomWalkRequest* req,
+                          std::vector<int64_t>* ret_walks,
+                          std::vector<int64_t>* ret_nbrs,
+                          std::vector<int32_t>* ret_degrees) {
+    int32_t batch_size = req->BatchSize();
+    auto src_ids = req->GetSrcIds();
+
+    const std::string& edge_type = req->Type();
+    Graph* graph = graph_store_->GetGraph(edge_type);
+    auto storage = graph->GetLocalStorage();
+
+    auto parent_ids = req->GetParentIds();
+    auto parent_neighbor_ids = req->GetParentNeighborIds();
+    auto parent_neighbor_indices = req->GetParentNeighborIndices();
+
+    float biased_weights[GLOBAL_FLAG(DefaultFullNbrNum)];
+    int32_t cursor = 0;
+
+    for (int32_t i = 0; i < batch_size; ++i) {
+      auto src_id = src_ids[i];
+      auto parent_id = parent_ids[i];
+      auto neighbor_ids = storage->GetNeighbors(src_id);
+
+      auto nbr_count = neighbor_ids.Size();
+      if (nbr_count == 0) {
+        ret_walks->push_back(GLOBAL_FLAG(DefaultNeighborId));
+      } else {
+        auto indice = WeightedRandomWalkKernel(
+          src_id, parent_id, parent_neighbor_ids,
+          cursor, parent_neighbor_indices[i],
+          req->P(), req->Q(), storage, biased_weights);
+        cursor += parent_neighbor_indices[i];
+
+        ret_walks->push_back(neighbor_ids[indice]);
+      }
+
+      int32_t degree = std::min(nbr_count, GLOBAL_FLAG(DefaultFullNbrNum));
+      if (degree > 0) {
+        std::copy_n(neighbor_ids.data(), degree, std::back_inserter(*ret_nbrs));
+      }
+      ret_degrees->push_back(degree);
+    }
+  }
+
+  int32_t WeightedRandomWalkKernel(int64_t src_id,
+                                   int64_t parent_id,
+                                   const int64_t* parent_neighbor_ids,
+                                   int32_t start,
+                                   int32_t size,
+                                   float p,
+                                   float q,
+                                   io::GraphStorage* storage,
+                                   float* biased_weights) {
+    auto neighbor_ids = storage->GetNeighbors(src_id);
+    auto edge_ids = storage->GetOutEdges(src_id);
+
+    size_t nbr_count = std::min(edge_ids.Size(), GLOBAL_FLAG(DefaultFullNbrNum));
+
+    for (size_t nbr_idx = 0; nbr_idx < nbr_count; ++nbr_idx) {
+      auto edge_weight = storage->GetEdgeWeight(edge_ids[nbr_idx]);
+      if (neighbor_ids[nbr_idx] == parent_id) {
+        biased_weights[nbr_idx] = edge_weight * 1.0 / (p + 1e-6);
+      } else {
+        int32_t idx = start;
+        for (; idx < start + size; ++idx) {
+          if (parent_neighbor_ids[idx] == neighbor_ids[nbr_idx]) {
+            biased_weights[nbr_idx] = edge_weight;
+            break;
+          }
+        }
+        if (idx == start + size) {
+          biased_weights[nbr_idx] = edge_weight * 1.0 / (q + 1e-6);
+        }
+      }
+    }
+
+    std::vector<float> weights(biased_weights, biased_weights + nbr_count);
+
+    AliasMethod am(&weights);
+    int32_t indice;
+    am.Sample(1, &indice);
+    return indice;
+  }
+};
+
+REGISTER_OPERATOR("RandomWalk", RandomWalk);
+
+}  // namespace op
+}  // namespace graphlearn
diff --git a/graphlearn/src/core/partition/hash_partitioner.h b/graphlearn/src/core/partition/hash_partitioner.h
index 7b4e9b95..46337f10 100644
--- a/graphlearn/src/core/partition/hash_partitioner.h
+++ b/graphlearn/src/core/partition/hash_partitioner.h
@@ -54,7 +54,24 @@ class HashPartitioner : public BasePartitioner<T> {
         if (it.first != kPartitionKey) {
           const Tensor* from = &(it.second);
           Tensor* target = &(part_req->tensors_[it.first]);
-          CopyToPartRequest(from, index, from->Size() / length, target);
+          size_t from_size = from->Size();
+          if (it.first == kSparseIds) {
+            std::vector<int32_t> cursors;
+            cursors.reserve(length + 1);
+            cursors.push_back(0);
+            auto& segs = req->tensors_.at(kSegments);
+            size_t size = segs.Size();
+            for (int32_t idx = 0; idx < length; ++idx) {
+              if (size < length) {
+                cursors.push_back(0);
+              } else {
+                cursors.push_back(segs.GetInt32(idx) + cursors[idx]);
+              }
+            }
+            CopyToPartRequest(from, index, cursors, target);
+          } else {
+            CopyToPartRequest(from, index, from->Size() / length, target);
+          }
         }
       }
     }
@@ -80,6 +97,31 @@ class HashPartitioner : public BasePartitioner<T> {
     return part_req;
   }
 
+   void CopyToPartRequest(const Tensor* from,
+                         int32_t index,
+                         const std::vector<int32_t>& dims,
+                         Tensor* target) {
+    DataType type = from->DType();
+
+#define CASE_COPY(Type)                           \
+  case k##Type:                                   \
+    for (int32_t i = dims[index]; i < dims[index + 1]; ++i) { \
+      target->Add##Type(from->Get##Type(i));      \
+    }                                             \
+    break
+
+    switch (type) {
+      CASE_COPY(Int64);
+      CASE_COPY(Int32);
+      CASE_COPY(Float);
+      CASE_COPY(Double);
+      CASE_COPY(String);
+      default:
+        break;
+    }
+#undef CASE_COPY
+  }
+
   void CopyToPartRequest(const Tensor* from,
                          int32_t index,
                          int32_t dim,
diff --git a/graphlearn/src/core/partition/stitcher.h b/graphlearn/src/core/partition/stitcher.h
index fb56eb51..28ecd8c6 100644
--- a/graphlearn/src/core/partition/stitcher.h
+++ b/graphlearn/src/core/partition/stitcher.h
@@ -122,11 +122,19 @@ class Stitcher {
         int32_t to_offset = incremental_degrees[offset];
         for (auto& it : tmp->tensors_) {
           if (it.first != kDegreeKey) {
-            CopyToResponse(&(it.second),
-                           from_offset,
-                           &(t->tensors_[it.first]),
-                           to_offset,
-                           dim);
+            if (tmp->IsSparse() && it.first == kNodeIds) {
+              CopyToResponse(&(it.second),
+                            i,
+                            &(t->tensors_[it.first]),
+                            offset,
+                            1);
+            } else {
+              CopyToResponse(&(it.second),
+                            from_offset,
+                            &(t->tensors_[it.first]),
+                            to_offset,
+                            dim);
+            }
           }
         }
         from_offset += dim;
@@ -171,8 +179,13 @@ class Stitcher {
 
     for (auto& it : tmp->tensors_) {
       if (it.first != kDegreeKey) {
-        ADD_TENSOR(t->tensors_, it.first, it.second.DType(), batch_degree);
-        t->tensors_[it.first].Resize(batch_degree);
+        if (tmp->IsSparse() && it.first == kNodeIds) {
+          ADD_TENSOR(t->tensors_, it.first, it.second.DType(), batch_size);
+          t->tensors_[it.first].Resize(batch_size);
+        } else {
+          ADD_TENSOR(t->tensors_, it.first, it.second.DType(), batch_degree);
+          t->tensors_[it.first].Resize(batch_degree);
+        }
       }
     }
 
diff --git a/graphlearn/src/include/client.h b/graphlearn/src/include/client.h
index a6d5bc0f..0948296e 100644
--- a/graphlearn/src/include/client.h
+++ b/graphlearn/src/include/client.h
@@ -20,6 +20,7 @@ limitations under the License.
 #include "include/constants.h"
 #include "include/dag_request.h"
 #include "include/graph_request.h"
+#include "include/random_walk_request.h"
 #include "include/sampling_request.h"
 #include "include/status.h"
 #include "include/subgraph_request.h"
@@ -49,6 +50,7 @@ class Client {
   DECLARE_METHOD(GetCount);
   DECLARE_METHOD(GetDegree);
   DECLARE_METHOD(GetStats);
+  DECLARE_METHOD(RandomWalk);
 
   Status RunOp(const OpRequest* request, OpResponse* response);
 
diff --git a/graphlearn/src/include/config.h b/graphlearn/src/include/config.h
index 08f5fbbc..da8d4543 100644
--- a/graphlearn/src/include/config.h
+++ b/graphlearn/src/include/config.h
@@ -96,6 +96,7 @@ DECLARE_INT32_GLOBAL_FLAG(TrackerMode)
 DECLARE_INT64_GLOBAL_FLAG(AverageNodeCount)
 DECLARE_INT64_GLOBAL_FLAG(AverageEdgeCount)
 DECLARE_INT64_GLOBAL_FLAG(DefaultNeighborId)
+DECLARE_INT32_GLOBAL_FLAG(DefaultFullNbrNum)
 DECLARE_INT64_GLOBAL_FLAG(DefaultIntAttribute)
 DECLARE_FLOAT_GLOBAL_FLAG(DefaultFloatAttribute)
 DECLARE_STRING_GLOBAL_FLAG(DefaultStringAttribute)
@@ -135,6 +136,7 @@ DECLARE_SET_INT32_GLOBAL_FLAG(TrackerMode)
 DECLARE_SET_INT64_GLOBAL_FLAG(AverageNodeCount)
 DECLARE_SET_INT64_GLOBAL_FLAG(AverageEdgeCount)
 DECLARE_SET_INT64_GLOBAL_FLAG(DefaultNeighborId)
+DECLARE_SET_INT32_GLOBAL_FLAG(DefaultFullNbrNum)
 DECLARE_SET_INT64_GLOBAL_FLAG(DefaultIntAttribute)
 DECLARE_SET_FLOAT_GLOBAL_FLAG(DefaultFloatAttribute)
 DECLARE_SET_STRING_GLOBAL_FLAG(DefaultStringAttribute)
diff --git a/graphlearn/src/include/constants.h b/graphlearn/src/include/constants.h
index 7a3f42e8..5d1b44eb 100644
--- a/graphlearn/src/include/constants.h
+++ b/graphlearn/src/include/constants.h
@@ -64,6 +64,7 @@ extern const char* kFilterIds;
 extern const char* kDegrees;
 extern const char* kEpoch;
 extern const char* kNodeFrom;
+extern const char* kSparseIds;
 
 enum SystemState {
   kBlank = 0,
diff --git a/graphlearn/src/include/random_walk_request.h b/graphlearn/src/include/random_walk_request.h
new file mode 100644
index 00000000..9ec806a6
--- /dev/null
+++ b/graphlearn/src/include/random_walk_request.h
@@ -0,0 +1,105 @@
+/* Copyright 2022 Alibaba Group Holding Limited. All Rights Reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+==============================================================================*/
+
+#ifndef GRAPHLEARN_INCLUDE_RANDOM_WALK_REQUEST_H_
+#define GRAPHLEARN_INCLUDE_RANDOM_WALK_REQUEST_H_
+
+#include <string>
+#include <unordered_map>
+#include <vector>
+#include "include/constants.h"
+#include "include/op_request.h"
+
+namespace graphlearn {
+
+class RandomWalkRequest : public OpRequest {
+public:
+  RandomWalkRequest();
+  RandomWalkRequest(const std::string& type, float p, float q, int32_t walk_len=1);
+  ~RandomWalkRequest() = default;
+
+  OpRequest* Clone() const override;
+
+  void Init(const Tensor::Map& params) override;
+  void Set(const Tensor::Map& tensors) override;
+  void Set(const int64_t* src_ids,
+           int32_t batch_size);
+  void Set(const int64_t* src_ids,
+           const int64_t* parent_ids,
+           int32_t batch_size,
+           const int64_t* parent_neighbor_ids,
+           const int32_t* parent_neighbor_indices,
+           int32_t total_count);
+
+  const bool IsDeepWalk() const;
+  const std::string& Type() const;
+  const float P() const;
+  const float Q() const;
+  const int32_t WalkLen() const;
+  int32_t BatchSize() const;
+  const int64_t* GetSrcIds() const;
+  const int64_t* GetParentIds() const;
+  const int64_t* GetParentNeighborIds() const;
+  const int32_t* GetParentNeighborIndices() const;
+
+protected:
+  void SetMembers() override;
+  // Params: kType, kPartitionKey, kSideInfo, kDistances
+  Tensor* src_ids_;  // kSrcIds
+  Tensor* parent_ids_;  // kNodeIds
+  Tensor* parent_neighbor_ids_;  // kSrcIds
+  Tensor* parent_neighbor_indices_;  // kNodeIds
+  int32_t total_neighbor_count_;
+};
+
+class RandomWalkResponse : public OpResponse {
+public:
+  RandomWalkResponse();
+  ~RandomWalkResponse() = default;
+
+  OpResponse* New() const override {
+    return new RandomWalkResponse;
+  }
+
+  void InitWalks(int32_t count); // init tensors
+  void InitNeighborIds(int32_t count);
+  void InitDegrees(int32_t count);
+
+  void SetBatchSize(int32_t batch_size);
+
+  void AppendWalks(const int64_t* walks, size_t size);
+  void AppendNeighborIds(const int64_t* ids, size_t size);
+  void AppendDegrees(const int32_t* degrees, size_t size);
+
+  void Swap(OpResponse& right) override;
+
+  const int64_t* GetWalks() const;
+  const int64_t* GetNeighborIds() const;
+  const int32_t* GetDegrees() const;
+
+
+protected:
+  void SetMembers() override;
+
+private:
+  Tensor* ids_;  // kNodeIds
+  Tensor* neighbors_;
+  Tensor* degrees_;
+};
+
+
+}  // namespace graphlearn
+
+#endif  // GRAPHLEARN_INCLUDE_RANDOM_WALK_REQUEST_H_
diff --git a/graphlearn/src/service/client.cc b/graphlearn/src/service/client.cc
index 11a06361..40ff7be2 100644
--- a/graphlearn/src/service/client.cc
+++ b/graphlearn/src/service/client.cc
@@ -50,6 +50,7 @@ DEFINE_METHOD(SubGraph);
 DEFINE_METHOD(GetCount);
 DEFINE_METHOD(GetDegree);
 DEFINE_METHOD(GetStats);
+DEFINE_METHOD(RandomWalk);
 
 #undef DEFINE_METHOD
 
diff --git a/graphlearn/src/service/constants.cc b/graphlearn/src/service/constants.cc
index f1a52117..2f4f6be4 100644
--- a/graphlearn/src/service/constants.cc
+++ b/graphlearn/src/service/constants.cc
@@ -61,5 +61,6 @@ const char* kFilterIds = "filt";
 const char* kDegrees = "dg";
 const char* kEpoch = "ep";
 const char* kNodeFrom = "nf";
+const char* kSparseIds = "sparse_ids";
 
 }  // namespace graphlearn
diff --git a/graphlearn/src/service/request/random_walk_request.cc b/graphlearn/src/service/request/random_walk_request.cc
new file mode 100644
index 00000000..d3701452
--- /dev/null
+++ b/graphlearn/src/service/request/random_walk_request.cc
@@ -0,0 +1,310 @@
+/* Copyright 2022 Alibaba Group Holding Limited. All Rights Reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+==============================================================================*/
+
+#include "include/random_walk_request.h"
+
+#include <cmath>
+#include <float.h>
+#include "include/constants.h"
+
+namespace graphlearn {
+
+namespace {
+  int32_t kReservedSize = 64;
+}  // anonymous namespace
+
+RandomWalkRequest::RandomWalkRequest()
+    : OpRequest(),
+      src_ids_(nullptr),
+      parent_ids_(nullptr),
+      parent_neighbor_ids_(nullptr),
+      parent_neighbor_indices_(nullptr) {
+}
+
+RandomWalkRequest::RandomWalkRequest(const std::string& type,
+                                     float p,
+                                     float q,
+                                     int32_t walk_len)
+    : OpRequest(),
+      src_ids_(nullptr),
+      parent_ids_(nullptr),
+      parent_neighbor_ids_(nullptr),
+      parent_neighbor_indices_(nullptr) {
+  params_.reserve(kReservedSize);
+  ADD_TENSOR(params_, kOpName, kString, 1);
+  params_[kOpName].AddString("RandomWalk");
+
+  ADD_TENSOR(params_, kEdgeType, kString, 1);
+  params_[kEdgeType].AddString(type);
+
+  ADD_TENSOR(params_, kPartitionKey, kString, 1);
+  params_[kPartitionKey].AddString(kSrcIds);
+
+  ADD_TENSOR(params_, kSideInfo, kFloat, 2);
+  params_[kSideInfo].AddFloat(p);
+  params_[kSideInfo].AddFloat(q);
+
+  ADD_TENSOR(params_, kDistances, kInt32, 1);
+  params_[kDistances].AddInt32(walk_len);
+
+  ADD_TENSOR(tensors_, kSrcIds, kInt64, kReservedSize);
+  src_ids_ = &(tensors_[kSrcIds]);  // size = batch_size
+
+  if (!IsDeepWalk()) {
+    ADD_TENSOR(tensors_, kNodeIds, kInt64, kReservedSize);
+    parent_ids_ = &(tensors_[kNodeIds]); // size  = batch size
+
+    ADD_TENSOR(tensors_, kSparseIds, kInt64, kReservedSize);
+    parent_neighbor_ids_ = &(tensors_[kSparseIds]);  // size = batch_size
+
+    ADD_TENSOR(tensors_, kSegments, kInt32, kReservedSize);
+    parent_neighbor_indices_ = &(tensors_[kSegments]); // size  = batch size
+  }
+}
+
+OpRequest* RandomWalkRequest::Clone() const {
+  RandomWalkRequest* req = new RandomWalkRequest(
+    Type(), P(), Q(), WalkLen());
+  return req;
+}
+
+void RandomWalkRequest::SetMembers() {
+  src_ids_ = &(tensors_[kSrcIds]);
+  if (!IsDeepWalk()) {
+    parent_ids_ = &(tensors_[kNodeIds]);
+    parent_neighbor_ids_ = &(tensors_[kSparseIds]);
+    parent_neighbor_indices_ = &(tensors_[kSegments]);
+  }
+}
+
+void RandomWalkRequest::Init(const Tensor::Map& params) {
+  params_.reserve(kReservedSize);
+  ADD_TENSOR(params_, kOpName, kString, 1);
+  params_[kOpName].AddString("RandomWalk");
+
+  ADD_TENSOR(params_, kEdgeType, kString, 1);
+  params_[kEdgeType].AddString(params.at(kEdgeType).GetString(0));
+
+  ADD_TENSOR(params_, kPartitionKey, kString, 1);
+  params_[kPartitionKey].AddString(kSrcIds);
+
+  ADD_TENSOR(params_, kSideInfo, kFloat, 2);
+  params_[kSideInfo].AddFloat(params.at(kSideInfo).GetFloat(0));
+  params_[kSideInfo].AddFloat(params.at(kSideInfo).GetFloat(1));
+
+  ADD_TENSOR(params_, kDistances, kInt32, 1);
+  params_[kDistances].AddInt32(params.at(kDistances).GetInt32(0));
+
+  ADD_TENSOR(tensors_, kSrcIds, kInt64, kReservedSize);
+  src_ids_ = &(tensors_[kSrcIds]);
+
+  if (!IsDeepWalk()) {
+    ADD_TENSOR(tensors_, kNodeIds, kInt64, kReservedSize);
+    parent_ids_ = &(tensors_[kNodeIds]);
+
+    ADD_TENSOR(tensors_, kSparseIds, kInt64, kReservedSize);
+    parent_neighbor_ids_ = &(tensors_[kSparseIds]);
+
+    ADD_TENSOR(tensors_, kSegments, kInt32, kReservedSize);
+    parent_neighbor_indices_ = &(tensors_[kSegments]); // size  = batch size
+  }
+}
+
+void RandomWalkRequest::Set(const Tensor::Map& tensors) {
+  auto src_ids = tensors.at(kSrcIds).GetInt64();
+  auto batch_size = tensors.at(kSrcIds).Size();
+  src_ids_->AddInt64(src_ids, src_ids + batch_size);
+
+  if (!IsDeepWalk()) {
+    parent_ids_->AddInt64(src_ids, src_ids + batch_size);
+    for (int32_t i = 0; i < batch_size; ++i) {
+      parent_neighbor_indices_->AddInt32(0);
+    }
+  }
+
+  // this->DisableShard();
+}
+
+void RandomWalkRequest::Set(const int64_t* src_ids,
+                            int32_t batch_size) {
+  src_ids_->AddInt64(src_ids, src_ids + batch_size);
+}
+
+void RandomWalkRequest::Set(const int64_t* src_ids,
+                            const int64_t* parent_ids,
+                            int32_t batch_size,
+                            const int64_t* parent_neighbor_ids,
+                            const int32_t* parent_neighbor_indices,
+                            int32_t total_count) {
+  src_ids_->AddInt64(src_ids, src_ids + batch_size);
+  parent_ids_->AddInt64(parent_ids, parent_ids + batch_size);
+  parent_neighbor_ids_->AddInt64(parent_neighbor_ids, parent_neighbor_ids + total_count);
+  parent_neighbor_indices_->AddInt32(parent_neighbor_indices, parent_neighbor_indices + batch_size);
+}
+
+int32_t RandomWalkRequest::BatchSize() const {
+  return src_ids_->Size();
+}
+
+const bool RandomWalkRequest::IsDeepWalk() const {
+  auto p = params_.at(kSideInfo).GetFloat(0);
+  auto q = params_.at(kSideInfo).GetFloat(1);
+  if (fabs(p - float(1.0)) < 32 * FLT_EPSILON &&
+      fabs(q - float(1.0)) < 32 * FLT_EPSILON) {
+    return true;
+  }
+  return false;
+}
+
+const std::string& RandomWalkRequest::Type() const {
+  return params_.at(kEdgeType).GetString(0);
+}
+
+const float RandomWalkRequest::P() const {
+  return params_.at(kSideInfo).GetFloat(0);
+}
+
+const float RandomWalkRequest::Q() const {
+  return params_.at(kSideInfo).GetFloat(1);
+}
+
+const int32_t RandomWalkRequest::WalkLen() const {
+  return params_.at(kDistances).GetInt32(0);
+}
+
+const int64_t* RandomWalkRequest::GetSrcIds() const {
+  if (src_ids_) {
+    return src_ids_->GetInt64();
+  } else {
+    return nullptr;
+  }
+}
+
+const int64_t* RandomWalkRequest::GetParentIds() const {
+  if (parent_ids_) {
+    return parent_ids_->GetInt64();
+  } else {
+    return nullptr;
+  }
+}
+
+const int64_t* RandomWalkRequest::GetParentNeighborIds() const {
+  if (parent_neighbor_ids_) {
+    return parent_neighbor_ids_->GetInt64();
+  } else {
+    return nullptr;
+  }
+}
+
+const int32_t* RandomWalkRequest::GetParentNeighborIndices() const {
+  if (parent_neighbor_indices_) {
+    return parent_neighbor_indices_->GetInt32();
+  } else {
+    return nullptr;
+  }
+}
+
+RandomWalkResponse::RandomWalkResponse()
+    : OpResponse(),
+      ids_(nullptr),
+      neighbors_(nullptr),
+      degrees_(nullptr) {
+}
+
+void RandomWalkResponse::Swap(OpResponse& right) {
+  OpResponse::Swap(right);
+  RandomWalkResponse& res = static_cast<RandomWalkResponse&>(right);
+  std::swap(ids_, res.ids_);
+  std::swap(neighbors_, res.neighbors_);
+  std::swap(degrees_, res.degrees_);
+}
+
+void RandomWalkResponse::SetMembers() {
+  ids_ = &(tensors_[kNodeIds]);
+
+  if (tensors_.find(kDstIds) != tensors_.end()) {
+    neighbors_ = &(tensors_[kDstIds]);
+  }
+  if (tensors_.find(kDegreeKey) != tensors_.end()) {
+    degrees_ = &(tensors_[kDegreeKey]);
+  }
+}
+
+void RandomWalkResponse::InitWalks(int32_t count) {
+  ADD_TENSOR(tensors_, kNodeIds, kInt64, count);
+  ids_ = &(tensors_[kNodeIds]);
+}
+
+void RandomWalkResponse::InitNeighborIds(int32_t count) {
+  if (IsSparse()) {
+    ADD_TENSOR(tensors_, kDstIds, kInt64, count);
+    neighbors_ = &(tensors_[kDstIds]);
+  }
+}
+
+void RandomWalkResponse::InitDegrees(int32_t count) {
+  if (IsSparse()) {
+    ADD_TENSOR(tensors_, kDegreeKey, kInt32, count);
+    degrees_ = &(tensors_[kDegreeKey]);
+  }
+}
+
+void RandomWalkResponse::SetBatchSize(int32_t batch_size) {
+  batch_size_ = batch_size;
+}
+
+void RandomWalkResponse::AppendWalks(const int64_t* walks, size_t size) {
+  ids_->AddInt64(walks, walks + size);
+}
+
+void RandomWalkResponse::AppendNeighborIds(const int64_t* ids, size_t size) {
+  if (IsSparse()) {
+    neighbors_->AddInt64(ids, ids + size);
+  }
+}
+
+void RandomWalkResponse::AppendDegrees(const int32_t* degrees, size_t size) {
+  if (IsSparse()) {
+    degrees_->AddInt32(degrees, degrees + size);
+  }
+}
+
+const int64_t* RandomWalkResponse:: GetWalks() const {
+  if (ids_) {
+    return ids_->GetInt64();
+  } else {
+    return nullptr;
+  }
+}
+
+const int64_t* RandomWalkResponse::GetNeighborIds() const {
+  if (neighbors_) {
+    return neighbors_->GetInt64();
+  } else {
+    return nullptr;
+  }
+}
+
+const int32_t* RandomWalkResponse::GetDegrees() const {
+  if (degrees_) {
+    return degrees_->GetInt32();
+  } else {
+    return nullptr;
+  }
+}
+
+REGISTER_REQUEST(RandomWalk, RandomWalkRequest, RandomWalkResponse);
+
+}  // namespace graphlearn
