diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/Aggregation.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/Aggregation.java
index cd85574bf..f34c22715 100644
--- a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/Aggregation.java
+++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/Aggregation.java
@@ -28,6 +28,7 @@ import org.springframework.data.mongodb.core.aggregation.ExposedFields.DirectFie
 import org.springframework.data.mongodb.core.aggregation.ExposedFields.ExposedField;
 import org.springframework.data.mongodb.core.aggregation.ExposedFields.FieldReference;
 import org.springframework.data.mongodb.core.aggregation.FieldsExposingAggregationOperation.InheritsFieldsAggregationOperation;
+import org.springframework.data.mongodb.core.aggregation.GraphLookupOperation.StartWithBuilder;
 import org.springframework.data.mongodb.core.aggregation.ReplaceRootOperation.ReplaceRootDocumentOperationBuilder;
 import org.springframework.data.mongodb.core.aggregation.ReplaceRootOperation.ReplaceRootOperationBuilder;
 import org.springframework.data.mongodb.core.aggregation.Fields.*;
@@ -318,6 +319,17 @@ public class Aggregation {
 		return new GroupOperation(fields);
 	}
 
+	/**
+	 * Creates a new {@link GraphLookupOperation.FromBuilder} to construct a {@link GraphLookupOperation} given
+	 * {@literal fromCollection}.
+	 *
+	 * @param fromCollection must not be {@literal null} or empty.
+	 * @return
+	 */
+	public static StartWithBuilder graphLookup(String fromCollection) {
+		return GraphLookupOperation.builder().from(fromCollection);
+	}
+
 	/**
 	 * Factory method to create a new {@link SortOperation} for the given {@link Sort}.
 	 *
diff --git a/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/GraphLookupOperation.java b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/GraphLookupOperation.java
new file mode 100644
index 000000000..1f7d24b48
--- /dev/null
+++ b/spring-data-mongodb/src/main/java/org/springframework/data/mongodb/core/aggregation/GraphLookupOperation.java
@@ -0,0 +1,365 @@
+/*
+ * Copyright 2016 the original author or authors.
+ *
+ * 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.
+ */
+package org.springframework.data.mongodb.core.aggregation;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import org.springframework.data.mongodb.core.aggregation.ExposedFields.ExposedField;
+import org.springframework.data.mongodb.core.aggregation.FieldsExposingAggregationOperation.InheritsFieldsAggregationOperation;
+import org.springframework.data.mongodb.core.query.CriteriaDefinition;
+import org.springframework.util.Assert;
+
+import com.mongodb.BasicDBList;
+import com.mongodb.BasicDBObject;
+import com.mongodb.DBObject;
+
+/**
+ * Encapsulates the aggregation framework {@code $graphLookup}-operation.
+ * <p>
+ * Performs a recursive search on a collection, with options for restricting the search by recursion depth and query
+ * filter.
+ * <p>
+ * We recommend to use the static factory method {@link Aggregation#graphLookup(String)} instead of creating instances
+ * of this class directly.
+ *
+ * @see http://docs.mongodb.org/manual/reference/aggregation/graphLookup/
+ * @author Mark Paluch
+ * @since 1.10
+ */
+public class GraphLookupOperation implements InheritsFieldsAggregationOperation {
+
+	private final String from;
+	private final List<Object> startWith;
+	private final Field connectFrom;
+	private final Field connectTo;
+	private final Field as;
+	private final Long maxDepth;
+	private final Field depthField;
+	private final CriteriaDefinition restrictSearchWithMatch;
+
+	private GraphLookupOperation(String from, List<Object> startWith, Field connectFrom, Field connectTo, Field as,
+			Long maxDepth, Field depthField, CriteriaDefinition restrictSearchWithMatch) {
+
+		this.from = from;
+		this.startWith = startWith;
+		this.connectFrom = connectFrom;
+		this.connectTo = connectTo;
+		this.as = as;
+		this.maxDepth = maxDepth;
+		this.depthField = depthField;
+		this.restrictSearchWithMatch = restrictSearchWithMatch;
+	}
+
+	/**
+	 * Creates a new {@link FromBuilder} to build {@link GraphLookupOperation}.
+	 * 
+	 * @return a new {@link FromBuilder}.
+	 */
+	public static FromBuilder builder() {
+		return new GraphLookupOperationFromBuilder();
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.aggregation.AggregationOperation#toDBObject(org.springframework.data.mongodb.core.aggregation.AggregationOperationContext)
+	 */
+	@Override
+	public DBObject toDBObject(AggregationOperationContext context) {
+
+		DBObject graphLookup = new BasicDBObject();
+
+		graphLookup.put("from", from);
+
+		BasicDBList list = new BasicDBList();
+
+		for (Object startWithElement : startWith) {
+
+			if (startWithElement instanceof AggregationExpression) {
+				list.add(((AggregationExpression) startWithElement).toDbObject(context));
+			}
+
+			if (startWithElement instanceof Field) {
+				list.add(context.getReference((Field) startWithElement).toString());
+			}
+		}
+
+		if (list.size() == 1) {
+			graphLookup.put("startWith", list.get(0));
+		} else {
+			graphLookup.put("startWith", list);
+		}
+
+		graphLookup.put("connectFromField", connectFrom.getName());
+		graphLookup.put("connectToField", connectTo.getName());
+		graphLookup.put("as", as.getName());
+
+		if (maxDepth != null) {
+			graphLookup.put("maxDepth", maxDepth);
+		}
+
+		if (depthField != null) {
+			graphLookup.put("depthField", depthField.getName());
+		}
+
+		if (restrictSearchWithMatch != null) {
+			graphLookup.put("restrictSearchWithMatch", context.getMappedObject(restrictSearchWithMatch.getCriteriaObject()));
+		}
+
+		return new BasicDBObject("$graphLookup", graphLookup);
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.mongodb.core.aggregation.FieldsExposingAggregationOperation#getFields()
+	 */
+	@Override
+	public ExposedFields getFields() {
+		return ExposedFields.from(new ExposedField(as, true));
+	}
+
+	/**
+	 * @author Mark Paluch
+	 */
+	public interface FromBuilder {
+
+		/**
+		 * Set the {@literal collectionName} to apply the {@code $graphLookup} to.
+		 *
+		 * @param collectionName must not be {@literal null} or empty.
+		 * @return
+		 */
+		StartWithBuilder from(String collectionName);
+	}
+
+	/**
+	 * @author Mark Paluch
+	 */
+	public interface StartWithBuilder {
+
+		/**
+		 * Set the startWith {@literal fieldReferences} to apply the {@code $graphLookup} to.
+		 *
+		 * @param fieldReferences must not be {@literal null}.
+		 * @return
+		 */
+		ConnectFromBuilder startWith(String... fieldReferences);
+
+		/**
+		 * Set the startWith {@literal expressions} to apply the {@code $graphLookup} to.
+		 *
+		 * @param expressions must not be {@literal null}.
+		 * @return
+		 */
+		ConnectFromBuilder startWith(AggregationExpression... expressions);
+	}
+
+	/**
+	 * @author Mark Paluch
+	 */
+	public interface ConnectFromBuilder {
+
+		/**
+		 * Set the connectFrom {@literal fieldName} to apply the {@code $graphLookup} to.
+		 *
+		 * @param fieldName must not be {@literal null} or empty.
+		 * @return
+		 */
+		ConnectToBuilder connectFrom(String fieldName);
+	}
+
+	/**
+	 * @author Mark Paluch
+	 */
+	public interface ConnectToBuilder {
+
+		/**
+		 * Set the connectTo {@literal fieldName} to apply the {@code $graphLookup} to.
+		 *
+		 * @param fieldName must not be {@literal null} or empty.
+		 * @return
+		 */
+		GraphLookupOperationBuilder connectTo(String fieldName);
+	}
+
+	/**
+	 * Builder to build the initial {@link GraphLookupOperationBuilder} that configures the initial mandatory set of
+	 * {@link GraphLookupOperation} properties.
+	 * 
+	 * @author Mark Paluch
+	 */
+	static final class GraphLookupOperationFromBuilder
+			implements FromBuilder, StartWithBuilder, ConnectFromBuilder, ConnectToBuilder {
+
+		private String from;
+		private List<? extends Object> startWith;
+		private String connectFrom;
+
+		/* (non-Javadoc)
+		 * @see org.springframework.data.mongodb.core.aggregation.GraphLookupOperation.FromBuilder#from(java.lang.String)
+		 */
+		@Override
+		public StartWithBuilder from(String collectionName) {
+
+			Assert.hasText(collectionName, "CollectionName must not be null or empty!");
+
+			this.from = collectionName;
+
+			return this;
+		}
+
+		/* (non-Javadoc)
+		 * @see org.springframework.data.mongodb.core.aggregation.GraphLookupOperation.StartWithBuilder#startWith(java.lang.String[])
+		 */
+		@Override
+		public ConnectFromBuilder startWith(String... fieldReferences) {
+
+			Assert.notNull(fieldReferences, "FieldReferences must not be null!");
+			Assert.noNullElements(fieldReferences, "FieldReferences must not contain null elements!");
+
+			List<Object> fields = new ArrayList<Object>(fieldReferences.length);
+
+			for (String fieldReference : fieldReferences) {
+				fields.add(Fields.field(fieldReference));
+			}
+
+			this.startWith = fields;
+
+			return this;
+		}
+
+		/* (non-Javadoc)
+		 * @see org.springframework.data.mongodb.core.aggregation.GraphLookupOperation.StartWithBuilder#startWith(org.springframework.data.mongodb.core.aggregation.AggregationExpression[])
+		 */
+		@Override
+		public ConnectFromBuilder startWith(AggregationExpression... expressions) {
+
+			Assert.notNull(expressions, "AggregationExpressions must not be null!");
+			Assert.noNullElements(expressions, "AggregationExpressions must not contain null elements!");
+
+			this.startWith = Arrays.asList(expressions);
+
+			return this;
+		}
+
+		/* (non-Javadoc)
+		 * @see org.springframework.data.mongodb.core.aggregation.GraphLookupOperation.ConnectFromBuilder#connectFrom(java.lang.String)
+		 */
+		@Override
+		public ConnectToBuilder connectFrom(String fieldName) {
+
+			Assert.hasText(fieldName, "ConnectFrom must not be null or empty!");
+
+			this.connectFrom = fieldName;
+
+			return this;
+		}
+
+		/* (non-Javadoc)
+		 * @see org.springframework.data.mongodb.core.aggregation.GraphLookupOperation.ConnectToBuilder#connectTo(java.lang.String)
+		 */
+		@Override
+		public GraphLookupOperationBuilder connectTo(String fieldName) {
+
+			Assert.hasText(fieldName, "ConnectTo must not be null or empty!");
+
+			return new GraphLookupOperationBuilder(from, startWith, connectFrom, fieldName);
+		}
+	}
+
+	/**
+	 * @author Mark Paluch
+	 */
+	static final class GraphLookupOperationBuilder {
+
+		private final String from;
+		private final List<Object> startWith;
+		private final Field connectFrom;
+		private final Field connectTo;
+		private Long maxDepth;
+		private Field depthField;
+		private CriteriaDefinition restrictSearchWithMatch;
+
+		protected GraphLookupOperationBuilder(String from, List<? extends Object> startWith, String connectFrom,
+				String connectTo) {
+
+			this.from = from;
+			this.startWith = new ArrayList<Object>(startWith);
+			this.connectFrom = Fields.field(connectFrom);
+			this.connectTo = Fields.field(connectTo);
+		}
+
+		/**
+		 * Limit the number of recursions.
+		 * 
+		 * @param numberOfRecursions must be greater or equal to zero.
+		 * @return
+		 */
+		public GraphLookupOperationBuilder maxDepth(long numberOfRecursions) {
+
+			Assert.isTrue(numberOfRecursions >= 0, "Max depth must be >= 0!");
+
+			this.maxDepth = numberOfRecursions;
+
+			return this;
+		}
+
+		/**
+		 * Add a depth field {@literal fieldName} to each traversed document in the search path.
+		 * 
+		 * @param fieldName must not be {@literal null} or empty.
+		 * @return
+		 */
+		public GraphLookupOperationBuilder depthField(String fieldName) {
+
+			Assert.hasText(fieldName, "Depth field name must not be null or empty!");
+
+			this.depthField = Fields.field(fieldName);
+
+			return this;
+		}
+
+		/**
+		 * Add a query specifying conditions to the recursive search.
+		 * 
+		 * @param criteriaDefinition must not be {@literal null}.
+		 * @return
+		 */
+		public GraphLookupOperationBuilder restrict(CriteriaDefinition criteriaDefinition) {
+
+			Assert.notNull(criteriaDefinition, "CriteriaDefinition must not be null!");
+
+			this.restrictSearchWithMatch = criteriaDefinition;
+
+			return this;
+		}
+
+		/**
+		 * Set the name of the array field added to each output document and return the final {@link GraphLookupOperation}.
+		 * Contains the documents traversed in the {@literal $graphLookup} stage to reach the document.
+		 * 
+		 * @param fieldName must not be {@literal null} or empty.
+		 * @return the final {@link GraphLookupOperation}.
+		 */
+		public GraphLookupOperation as(String fieldName) {
+
+			Assert.hasText(fieldName, "As field name must not be null or empty!");
+
+			return new GraphLookupOperation(from, startWith, connectFrom, connectTo, Fields.field(fieldName), maxDepth,
+					depthField, restrictSearchWithMatch);
+		}
+	}
+}
diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/aggregation/AggregationTests.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/aggregation/AggregationTests.java
index 6cfc04b9f..2d0bb7750 100644
--- a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/aggregation/AggregationTests.java
+++ b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/aggregation/AggregationTests.java
@@ -72,6 +72,7 @@ import org.springframework.data.util.Version;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
+import com.mongodb.BasicDBList;
 import com.mongodb.BasicDBObject;
 import com.mongodb.BasicDBObjectBuilder;
 import com.mongodb.CommandResult;
@@ -146,6 +147,7 @@ public class AggregationTests {
 		mongoTemplate.dropCollection(InventoryItem.class);
 		mongoTemplate.dropCollection(Sales.class);
 		mongoTemplate.dropCollection(Sales2.class);
+		mongoTemplate.dropCollection(Employee.class);
 	}
 
 	/**
@@ -1631,6 +1633,40 @@ public class AggregationTests {
 						new BasicDBObjectBuilder().add("_id", "2").add("finalTotal", 10.25D).get()));
 	}
 
+	/**
+	 * @see DATAMONGO-1551
+	 */
+	@Test
+	public void graphLookupShouldBeAppliedCorrectly() {
+
+		assumeTrue(mongoVersion.isGreaterThanOrEqualTo(THREE_DOT_FOUR));
+
+		Employee em1 = Employee.builder().id(1).name("Dev").build();
+		Employee em2 = Employee.builder().id(2).name("Eliot").reportsTo("Dev").build();
+		Employee em4 = Employee.builder().id(4).name("Andrew").reportsTo("Eliot").build();
+
+		mongoTemplate.insert(Arrays.asList(em1, em2, em4), Employee.class);
+
+		TypedAggregation<Employee> agg = Aggregation.newAggregation(Employee.class,
+				match(Criteria.where("name").is("Andrew")), //
+				Aggregation.graphLookup("employee") //
+						.startWith("reportsTo") //
+						.connectFrom("reportsTo") //
+						.connectTo("name") //
+						.depthField("depth") //
+						.maxDepth(5) //
+						.as("reportingHierarchy"));
+
+		AggregationResults<DBObject> result = mongoTemplate.aggregate(agg, DBObject.class);
+
+		DBObject object = result.getUniqueMappedResult();
+		BasicDBList list = (BasicDBList) object.get("reportingHierarchy");
+
+		assertThat(object, isBsonObject().containing("reportingHierarchy", List.class));
+		assertThat((DBObject) list.get(0), isBsonObject().containing("name", "Dev").containing("depth", 1L));
+		assertThat((DBObject) list.get(1), isBsonObject().containing("name", "Eliot").containing("depth", 0L));
+	}
+
 	private void createUsersWithReferencedPersons() {
 
 		mongoTemplate.dropCollection(User.class);
@@ -1873,7 +1909,7 @@ public class AggregationTests {
 	}
 
 	/**
-	 * @DATAMONGO-1491
+	 * @see DATAMONGO-1491
 	 */
 	@lombok.Data
 	@Builder
@@ -1884,7 +1920,7 @@ public class AggregationTests {
 	}
 
 	/**
-	 * @DATAMONGO-1491
+	 * @see DATAMONGO-1491
 	 */
 	@lombok.Data
 	@Builder
@@ -1897,7 +1933,7 @@ public class AggregationTests {
 	}
 
 	/**
-	 * @DATAMONGO-1538
+	 * @see DATAMONGO-1538
 	 */
 	@lombok.Data
 	@Builder
@@ -1908,4 +1944,16 @@ public class AggregationTests {
 		Float tax;
 		boolean applyDiscount;
 	}
+
+	/**
+	 * @see DATAMONGO-1551
+	 */
+	@lombok.Data
+	@Builder
+	static class Employee {
+
+		int id;
+		String name;
+		String reportsTo;
+	}
 }
diff --git a/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/aggregation/GraphLookupOperationUnitTests.java b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/aggregation/GraphLookupOperationUnitTests.java
new file mode 100644
index 000000000..091e20106
--- /dev/null
+++ b/spring-data-mongodb/src/test/java/org/springframework/data/mongodb/core/aggregation/GraphLookupOperationUnitTests.java
@@ -0,0 +1,122 @@
+/*
+ * Copyright 2016 the original author or authors.
+ *
+ * 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.
+ */
+package org.springframework.data.mongodb.core.aggregation;
+
+import static org.hamcrest.core.Is.*;
+import static org.junit.Assert.*;
+import static org.springframework.data.mongodb.test.util.IsBsonObject.*;
+
+import org.junit.Test;
+import org.springframework.data.mongodb.core.aggregation.AggregationExpressions.Literal;
+import org.springframework.data.mongodb.core.query.Criteria;
+
+import com.mongodb.BasicDBObject;
+import com.mongodb.DBObject;
+import com.mongodb.util.JSON;
+
+/**
+ * Unit tests for {@link GraphLookupOperation}.
+ * 
+ * @author Mark Paluch
+ */
+public class GraphLookupOperationUnitTests {
+
+	/**
+	 * @see DATAMONGO-1551
+	 */
+	@Test(expected = IllegalArgumentException.class)
+	public void rejectsNullFromCollection() {
+		GraphLookupOperation.builder().from(null);
+	}
+
+	/**
+	 * @see DATAMONGO-1551
+	 */
+	@Test
+	public void shouldRenderCorrectly() {
+
+		GraphLookupOperation graphLookupOperation = GraphLookupOperation.builder() //
+				.from("employees") //
+				.startWith("reportsTo") //
+				.connectFrom("reportsTo") //
+				.connectTo("name") //
+				.depthField("depth") //
+				.maxDepth(42) //
+				.as("reportingHierarchy");
+
+		DBObject dbObject = graphLookupOperation.toDBObject(Aggregation.DEFAULT_CONTEXT);
+		assertThat(dbObject,
+				isBsonObject().containing("$graphLookup.depthField", "depth").containing("$graphLookup.maxDepth", 42L));
+	}
+
+	/**
+	 * @see DATAMONGO-1551
+	 */
+	@Test
+	public void shouldRenderCriteriaCorrectly() {
+
+		GraphLookupOperation graphLookupOperation = GraphLookupOperation.builder() //
+				.from("employees") //
+				.startWith("reportsTo") //
+				.connectFrom("reportsTo") //
+				.connectTo("name") //
+				.restrict(Criteria.where("key").is("value")) //
+				.as("reportingHierarchy");
+
+		DBObject dbObject = graphLookupOperation.toDBObject(Aggregation.DEFAULT_CONTEXT);
+		assertThat(dbObject,
+				isBsonObject().containing("$graphLookup.restrictSearchWithMatch", new BasicDBObject("key", "value")));
+	}
+
+	/**
+	 * @see DATAMONGO-1551
+	 */
+	@Test
+	public void shouldRenderArrayOfStartsWithCorrectly() {
+
+		GraphLookupOperation graphLookupOperation = GraphLookupOperation.builder() //
+				.from("employees") //
+				.startWith("reportsTo", "boss") //
+				.connectFrom("reportsTo") //
+				.connectTo("name") //
+				.as("reportingHierarchy");
+
+		DBObject dbObject = graphLookupOperation.toDBObject(Aggregation.DEFAULT_CONTEXT);
+
+		assertThat(dbObject,
+				is(JSON.parse("{ $graphLookup : { from: \"employees\", startWith: [\"$reportsTo\", \"$boss\"], "
+						+ "connectFromField: \"reportsTo\", connectToField: \"name\", as: \"reportingHierarchy\" } }")));
+	}
+
+	/**
+	 * @see DATAMONGO-1551
+	 */
+	@Test
+	public void shouldRenderStartWithAggregationExpressions() {
+
+		GraphLookupOperation graphLookupOperation = GraphLookupOperation.builder() //
+				.from("employees") //
+				.startWith(Literal.asLiteral("hello")) //
+				.connectFrom("reportsTo") //
+				.connectTo("name") //
+				.as("reportingHierarchy");
+
+		DBObject dbObject = graphLookupOperation.toDBObject(Aggregation.DEFAULT_CONTEXT);
+
+		assertThat(dbObject, is(JSON.parse("{ $graphLookup : { from: \"employees\", startWith: { $literal: \"hello\"}, "
+				+ "connectFromField: \"reportsTo\", connectToField: \"name\", as: \"reportingHierarchy\" } }")));
+	}
+}