diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/builder/JdbcBatchItemWriterBuilder.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/builder/JdbcBatchItemWriterBuilder.java
new file mode 100644
index 000000000..7b8ffc0ad
--- /dev/null
+++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/database/builder/JdbcBatchItemWriterBuilder.java
@@ -0,0 +1,199 @@
+/*
+ * 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.batch.item.database.builder;
+
+import java.math.BigInteger;
+import java.util.Map;
+import javax.sql.DataSource;
+
+import org.springframework.batch.item.database.BeanPropertyItemSqlParameterSourceProvider;
+import org.springframework.batch.item.database.ItemPreparedStatementSetter;
+import org.springframework.batch.item.database.ItemSqlParameterSourceProvider;
+import org.springframework.batch.item.database.JdbcBatchItemWriter;
+import org.springframework.batch.item.database.support.ColumnMapItemPreparedStatementSetter;
+import org.springframework.jdbc.core.namedparam.NamedParameterJdbcOperations;
+import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
+import org.springframework.util.Assert;
+
+/**
+ * A builder implementation for the {@link JdbcBatchItemWriter}.
+ *
+ * @author Michael Minella
+ * @since 4.0
+ * @see JdbcBatchItemWriter
+ */
+public class JdbcBatchItemWriterBuilder<T> {
+
+	private boolean assertUpdates = true;
+
+	private String sql;
+
+	private ItemPreparedStatementSetter<T> itemPreparedStatementSetter;
+
+	private ItemSqlParameterSourceProvider<T> itemSqlParameterSourceProvider;
+
+	private DataSource dataSource;
+
+	private NamedParameterJdbcOperations namedParameterJdbcTemplate;
+
+	private BigInteger mapped = new BigInteger("0");
+
+	/**
+	 * Configure the {@link DataSource} to be used.
+	 *
+	 * @param dataSource the DataSource
+	 * @return The current instance of the builder for chaining.
+	 * @see JdbcBatchItemWriter#setDataSource(DataSource)
+	 */
+	public JdbcBatchItemWriterBuilder<T> dataSource(DataSource dataSource) {
+		this.dataSource = dataSource;
+
+		return this;
+	}
+
+	/**
+	 * If set to true, confirms that every insert results in the update of at least one
+	 * row in the database.  Defaults to true.
+	 *
+	 * @param assertUpdates boolean indicator
+	 * @return The current instance of the builder for chaining
+	 * @see JdbcBatchItemWriter#setAssertUpdates(boolean)
+	 */
+	public JdbcBatchItemWriterBuilder<T> assertUpdates(boolean assertUpdates) {
+		this.assertUpdates = assertUpdates;
+
+		return this;
+	}
+
+	/**
+	 * Set the SQL statement to be used for each item's updates.  This is a required
+	 * field.
+	 *
+	 * @param sql SQL string
+	 * @return The current instance of the builder for chaining
+	 * @see JdbcBatchItemWriter#setSql(String)
+	 */
+	public JdbcBatchItemWriterBuilder<T> sql(String sql) {
+		this.sql = sql;
+
+		return this;
+	}
+
+	/**
+	 * Configures a {@link ItemPreparedStatementSetter} for use by the writer.  This
+	 * should only be used if {@link #columnMapped()} isn't called.
+	 *
+	 * @param itemPreparedStatementSetter The {@link ItemPreparedStatementSetter}
+	 * @return The current instance of the builder for chaining
+	 * @see JdbcBatchItemWriter#setItemPreparedStatementSetter(ItemPreparedStatementSetter)
+	 */
+	public JdbcBatchItemWriterBuilder<T> itemPreparedStatementSetter(ItemPreparedStatementSetter<T> itemPreparedStatementSetter) {
+		this.itemPreparedStatementSetter = itemPreparedStatementSetter;
+
+		return this;
+	}
+
+	/**
+	 * Configures a {@link ItemSqlParameterSourceProvider} for use by the writer.  This
+	 * should only be used if {@link #beanMapped()} isn't called.
+	 *
+	 * @param itemSqlParameterSourceProvider The {@link ItemSqlParameterSourceProvider}
+	 * @return The current instance of the builder for chaining
+	 * @see JdbcBatchItemWriter#setItemSqlParameterSourceProvider(ItemSqlParameterSourceProvider)
+	 */
+	public JdbcBatchItemWriterBuilder<T> itemSqlParameterSourceProvider(ItemSqlParameterSourceProvider<T> itemSqlParameterSourceProvider) {
+		this.itemSqlParameterSourceProvider = itemSqlParameterSourceProvider;
+
+		return this;
+	}
+
+	/**
+	 * The {@link NamedParameterJdbcOperations} instance to use.  If one isn't provided,
+	 * a {@link DataSource} is required.
+	 *
+	 * @param namedParameterJdbcOperations The template
+	 * @return The current instance of the builder for chaining
+	 */
+	public JdbcBatchItemWriterBuilder<T> namedParametersJdbcTemplate(NamedParameterJdbcOperations namedParameterJdbcOperations) {
+		this.namedParameterJdbcTemplate = namedParameterJdbcOperations;
+
+		return this;
+	}
+
+	/**
+	 * Creates a {@link ColumnMapItemPreparedStatementSetter} to be used as your
+	 * {@link ItemPreparedStatementSetter}.
+	 *
+	 * NOTE: The item type for this {@link org.springframework.batch.item.ItemWriter} must
+	 * be castable to <code>Map&lt;String,Object&gt;&gt;</code>.
+	 *
+	 * @return The current instance of the builder for chaining
+	 * @see ColumnMapItemPreparedStatementSetter
+	 */
+	public JdbcBatchItemWriterBuilder<T> columnMapped() {
+		this.mapped = this.mapped.setBit(0);
+
+		return this;
+	}
+
+	/**
+	 * Creates a {@link BeanPropertyItemSqlParameterSourceProvider} to be used as your
+	 * {@link ItemSqlParameterSourceProvider}.
+	 *
+	 * @return The current instance of the builder for chaining
+	 * @see BeanPropertyItemSqlParameterSourceProvider
+	 */
+	public JdbcBatchItemWriterBuilder<T> beanMapped() {
+		this.mapped = this.mapped.setBit(1);
+
+		return this;
+	}
+
+	/**
+	 * Validates configuration and builds the {@link JdbcBatchItemWriter}.
+	 *
+	 * @return a {@link JdbcBatchItemWriter}
+	 */
+	public JdbcBatchItemWriter<T> build() {
+		Assert.state(this.dataSource != null || this.namedParameterJdbcTemplate != null,
+				"Either a DataSource or a NamedParameterJdbcTemplate is required");
+
+		Assert.notNull(this.sql, "A SQL statement is required");
+		int mappedValue = this.mapped.intValue();
+		Assert.state(mappedValue != 3,
+				"Either an item can be mapped via db column or via bean spec, can't be both");
+
+		JdbcBatchItemWriter<T> writer = new JdbcBatchItemWriter<>();
+		writer.setSql(this.sql);
+		writer.setAssertUpdates(this.assertUpdates);
+		writer.setItemSqlParameterSourceProvider(this.itemSqlParameterSourceProvider);
+		writer.setItemPreparedStatementSetter(this.itemPreparedStatementSetter);
+
+		if(mappedValue == 1) {
+			((JdbcBatchItemWriter<Map<String,Object>>)writer).setItemPreparedStatementSetter(new ColumnMapItemPreparedStatementSetter());
+		} else if(mappedValue == 2) {
+			writer.setItemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<>());
+		}
+
+		if(this.dataSource != null) {
+			this.namedParameterJdbcTemplate = new NamedParameterJdbcTemplate(this.dataSource);
+		}
+
+		writer.setJdbcTemplate(this.namedParameterJdbcTemplate);
+
+		return writer;
+	}
+}
diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/builder/JdbcBatchItemWriterBuilderTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/builder/JdbcBatchItemWriterBuilderTests.java
new file mode 100644
index 000000000..e391eb076
--- /dev/null
+++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/database/builder/JdbcBatchItemWriterBuilderTests.java
@@ -0,0 +1,330 @@
+/*
+ * 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.batch.item.database.builder;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import javax.sql.DataSource;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import org.springframework.batch.item.database.JdbcBatchItemWriter;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.context.annotation.AnnotationConfigApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.core.io.ByteArrayResource;
+import org.springframework.core.io.Resource;
+import org.springframework.dao.EmptyResultDataAccessException;
+import org.springframework.jdbc.core.JdbcOperations;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.core.namedparam.MapSqlParameterSource;
+import org.springframework.jdbc.core.namedparam.NamedParameterJdbcOperations;
+import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
+import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseFactory;
+import org.springframework.jdbc.datasource.init.DataSourceInitializer;
+import org.springframework.jdbc.datasource.init.ResourceDatabasePopulator;
+import org.springframework.test.util.ReflectionTestUtils;
+
+import static junit.framework.Assert.assertTrue;
+import static junit.framework.TestCase.assertEquals;
+import static org.junit.Assert.fail;
+
+/**
+ * @author Michael Minella
+ */
+public class JdbcBatchItemWriterBuilderTests {
+
+	private DataSource dataSource;
+
+	private ConfigurableApplicationContext context;
+
+	@Before
+	public void setUp() {
+		this.context = new AnnotationConfigApplicationContext(TestDataSourceConfiguration.class);
+		this.dataSource = (DataSource) context.getBean("dataSource");
+	}
+
+	@After
+	public void tearDown() {
+		if(this.context != null) {
+			this.context.close();
+		}
+	}
+
+	@Test
+	public void testBasicMap() throws Exception {
+		JdbcBatchItemWriter<Map<String, Object>> writer = new JdbcBatchItemWriterBuilder<Map<String, Object>>()
+				.columnMapped()
+				.dataSource(this.dataSource)
+				.sql("INSERT INTO FOO (first, second, third) VALUES (:first, :second, :third)")
+				.build();
+
+		writer.afterPropertiesSet();
+
+		List<Map<String, Object>> items = buildMapItems();
+		writer.write(items);
+
+		verifyWrite();
+	}
+
+	private void verifyWrite() {
+		verifyRow(1, "two", "three");
+		verifyRow(4, "five", "six");
+		verifyRow(7, "eight", "nine");
+	}
+
+	@Test
+	public void testCustomJdbcTemplate() throws Exception {
+		NamedParameterJdbcOperations template = new NamedParameterJdbcTemplate(this.dataSource);
+
+		JdbcBatchItemWriter<Map<String, Object>> writer = new JdbcBatchItemWriterBuilder<Map<String, Object>>()
+				.columnMapped()
+				.namedParametersJdbcTemplate(template)
+				.sql("INSERT INTO FOO (first, second, third) VALUES (:first, :second, :third)")
+				.build();
+
+		writer.afterPropertiesSet();
+
+		List<Map<String, Object>> items = buildMapItems();
+		writer.write(items);
+
+		verifyWrite();
+
+		Object usedTemplate = ReflectionTestUtils.getField(writer, "namedParameterJdbcTemplate");
+		assertTrue(template == usedTemplate);
+	}
+
+	@Test
+	public void testBasicPojo() throws Exception {
+		JdbcBatchItemWriter<Foo> writer = new JdbcBatchItemWriterBuilder<Foo>()
+				.beanMapped()
+				.dataSource(this.dataSource)
+				.sql("INSERT INTO FOO (first, second, third) VALUES (:first, :second, :third)")
+				.build();
+
+		writer.afterPropertiesSet();
+
+		List<Foo> items = new ArrayList<>(3);
+
+		items.add(new Foo(1, "two", "three"));
+		items.add(new Foo(4, "five", "six"));
+		items.add(new Foo(7, "eight", "nine"));
+
+		writer.write(items);
+
+		verifyWrite();
+	}
+
+	@Test(expected = EmptyResultDataAccessException.class)
+	public void testAssertUpdates() throws Exception {
+		JdbcBatchItemWriter<Foo> writer = new JdbcBatchItemWriterBuilder<Foo>()
+				.beanMapped()
+				.dataSource(this.dataSource)
+				.sql("UPDATE FOO SET second = :second, third = :third WHERE first = :first")
+				.assertUpdates(true)
+				.build();
+
+		writer.afterPropertiesSet();
+
+		List<Foo> items = new ArrayList<>(1);
+
+		items.add(new Foo(1, "two", "three"));
+
+		writer.write(items);
+	}
+
+	@Test
+	public void testCustomPreparedStatementSetter() throws Exception {
+		JdbcBatchItemWriter<Map<String, Object>> writer = new JdbcBatchItemWriterBuilder<Map<String, Object>>()
+				.itemPreparedStatementSetter((item, ps) -> {
+					ps.setInt(0, (int) item.get("first"));
+					ps.setString(1, (String) item.get("second"));
+					ps.setString(2, (String) item.get("third"));
+				})
+				.dataSource(this.dataSource)
+				.sql("INSERT INTO FOO (first, second, third) VALUES (:first, :second, :third)")
+				.build();
+
+		writer.afterPropertiesSet();
+
+		List<Map<String, Object>> items = buildMapItems();
+		writer.write(items);
+
+		verifyWrite();
+	}
+
+	@Test
+	public void testCustomPSqlParameterSourceProvider() throws Exception {
+		JdbcBatchItemWriter<Map<String, Object>> writer = new JdbcBatchItemWriterBuilder<Map<String, Object>>()
+				.itemSqlParameterSourceProvider(MapSqlParameterSource::new)
+				.dataSource(this.dataSource)
+				.sql("INSERT INTO FOO (first, second, third) VALUES (:first, :second, :third)")
+				.build();
+
+		writer.afterPropertiesSet();
+
+		List<Map<String, Object>> items = buildMapItems();
+		writer.write(items);
+
+		verifyWrite();
+	}
+
+	@Test
+	public void testBuildAssertions() {
+		try {
+			new JdbcBatchItemWriterBuilder<Map<String, Object>>()
+					.itemSqlParameterSourceProvider(MapSqlParameterSource::new)
+					.build();
+		}
+		catch (IllegalStateException ise) {
+			assertEquals("Either a DataSource or a NamedParameterJdbcTemplate is required",
+					ise.getMessage());
+		}
+		catch (Exception e) {
+			fail("Incorrect exception was thrown when missing DataSource and JdbcTemplate: " +
+					e.getMessage());
+		}
+
+		try {
+			new JdbcBatchItemWriterBuilder<Map<String, Object>>()
+					.itemSqlParameterSourceProvider(MapSqlParameterSource::new)
+					.dataSource(this.dataSource)
+					.build();
+		}
+		catch (IllegalArgumentException ise) {
+			assertEquals("A SQL statement is required", ise.getMessage());
+		}
+		catch (Exception e) {
+			fail("Incorrect exception was thrown when testing missing SQL: " +
+					e);
+		}
+
+		try {
+			new JdbcBatchItemWriterBuilder<Map<String, Object>>()
+					.dataSource(this.dataSource)
+					.sql("INSERT INTO FOO VALUES (?, ?, ?)")
+					.columnMapped()
+					.beanMapped()
+					.build();
+		}
+		catch (IllegalStateException ise) {
+			assertEquals("Either an item can be mapped via db column or via bean spec, can't be both",
+					ise.getMessage());
+		}
+		catch (Exception e) {
+			fail("Incorrect exception was thrown both mapping types are used" +
+					e.getMessage());
+		}
+	}
+
+	private List<Map<String, Object>> buildMapItems() {
+		List<Map<String, Object>> items = new ArrayList<>(3);
+
+		Map<String, Object> item = new HashMap<>(3);
+		item.put("first", 1);
+		item.put("second", "two");
+		item.put("third", "three");
+		items.add(item);
+
+		item = new HashMap<>(3);
+		item.put("first", 4);
+		item.put("second", "five");
+		item.put("third", "six");
+		items.add(item);
+
+		item = new HashMap<>(3);
+		item.put("first", 7);
+		item.put("second", "eight");
+		item.put("third", "nine");
+		items.add(item);
+		return items;
+	}
+
+	private void verifyRow(int i, String i1, String nine) {
+		JdbcOperations template = new JdbcTemplate(this.dataSource);
+
+		assertEquals(1, (int) template.queryForObject(
+				"select count(*) from foo where first = ? and second = ? and third = ?",
+				new Object[] {i, i1, nine}, Integer.class));
+	}
+
+	public static class Foo {
+		private int first;
+		private String second;
+		private String third;
+
+		public Foo(int first, String second, String third) {
+			this.first = first;
+			this.second = second;
+			this.third = third;
+		}
+
+		public int getFirst() {
+			return first;
+		}
+
+		public void setFirst(int first) {
+			this.first = first;
+		}
+
+		public String getSecond() {
+			return second;
+		}
+
+		public void setSecond(String second) {
+			this.second = second;
+		}
+
+		public String getThird() {
+			return third;
+		}
+
+		public void setThird(String third) {
+			this.third = third;
+		}
+	}
+
+	@Configuration
+	public static class TestDataSourceConfiguration {
+
+		private static final String CREATE_SQL = "CREATE TABLE FOO  (\n" +
+				"\tID BIGINT IDENTITY NOT NULL PRIMARY KEY ,\n" +
+				"\tFIRST BIGINT ,\n" +
+				"\tSECOND VARCHAR(5) NOT NULL,\n" +
+				"\tTHIRD VARCHAR(5) NOT NULL) ;";
+
+		@Bean
+		public DataSource dataSource() {
+			return new EmbeddedDatabaseFactory().getDatabase();
+		}
+
+		@Bean
+		public DataSourceInitializer initializer(DataSource dataSource) {
+			DataSourceInitializer dataSourceInitializer = new DataSourceInitializer();
+			dataSourceInitializer.setDataSource(dataSource);
+
+			Resource create = new ByteArrayResource(CREATE_SQL.getBytes());
+			dataSourceInitializer.setDatabasePopulator(new ResourceDatabasePopulator(create));
+
+			return dataSourceInitializer;
+		}
+	}
+}