diff --git a/table/annotation-processor/src/main/java/com/riiablo/table/annotation/CodeGenerator.java b/table/annotation-processor/src/main/java/com/riiablo/table/annotation/CodeGenerator.java
index 4d8c6a9f..8ed0c76e 100644
--- a/table/annotation-processor/src/main/java/com/riiablo/table/annotation/CodeGenerator.java
+++ b/table/annotation-processor/src/main/java/com/riiablo/table/annotation/CodeGenerator.java
@@ -1,6 +1,7 @@
 package com.riiablo.table.annotation;
 
 import com.squareup.javapoet.AnnotationSpec;
+import com.squareup.javapoet.ClassName;
 import com.squareup.javapoet.JavaFile;
 import com.squareup.javapoet.TypeSpec;
 import java.util.Date;
@@ -17,16 +18,16 @@ abstract class CodeGenerator {
     this.packageName = packageName;
   }
 
-  abstract String formatName(SchemaElement schemaElement);
+  abstract ClassName formatName(String packageName, SchemaElement schemaElement);
 
   JavaFile generate(SchemaElement schemaElement) {
-    TypeSpec.Builder serializerTypeSpec
+    TypeSpec.Builder typeSpecBuilder
         =newTypeSpec(schemaElement)
         .addAnnotation(newGenerated(schemaElement.element.getQualifiedName().toString()))
         ;
 
     return JavaFile
-        .builder(packageName, serializerTypeSpec.build())
+        .builder(packageName, typeSpecBuilder.build())
         .skipJavaLangImports(true)
         .addFileComment(
             "automatically generated by $L, do not modify",
@@ -35,7 +36,7 @@ abstract class CodeGenerator {
   }
 
   TypeSpec.Builder newTypeSpec(SchemaElement schemaElement) {
-    return TypeSpec.classBuilder(formatName(schemaElement));
+    return TypeSpec.classBuilder(formatName(packageName, schemaElement));
   }
 
   AnnotationSpec newGenerated(String comments) {
diff --git a/table/annotation-processor/src/main/java/com/riiablo/table/annotation/SchemaElement.java b/table/annotation-processor/src/main/java/com/riiablo/table/annotation/SchemaElement.java
index da624302..47e73f85 100644
--- a/table/annotation-processor/src/main/java/com/riiablo/table/annotation/SchemaElement.java
+++ b/table/annotation-processor/src/main/java/com/riiablo/table/annotation/SchemaElement.java
@@ -127,6 +127,8 @@ final class SchemaElement {
   final FieldElement primaryKeyFieldElement;
   final Collection<FieldElement> fields;
 
+  ClassName serializerClassName;
+
   SchemaElement(
       Schema annotation,
       TypeElement element,
@@ -140,6 +142,9 @@ final class SchemaElement {
     this.serializerElement = serializerElement;
     this.primaryKeyFieldElement = primaryKeyFieldElement;
     this.fields = fields;
+    if (serializerElement.serializerImplElement != null) {
+      serializerClassName = ClassName.get(serializerElement.serializerImplElement);
+    }
   }
 
   @Override
@@ -148,6 +153,7 @@ final class SchemaElement {
         .append("element", element)
         .append("tableElement", tableElement)
         .append("serializerElement", serializerElement)
+        .append("serializerClassName", serializerClassName)
         .append("primaryKeyFieldElement", primaryKeyFieldElement)
         .toString();
   }
diff --git a/table/annotation-processor/src/main/java/com/riiablo/table/annotation/SchemaProcessor.java b/table/annotation-processor/src/main/java/com/riiablo/table/annotation/SchemaProcessor.java
index c7cf7ea6..7280fcbf 100644
--- a/table/annotation-processor/src/main/java/com/riiablo/table/annotation/SchemaProcessor.java
+++ b/table/annotation-processor/src/main/java/com/riiablo/table/annotation/SchemaProcessor.java
@@ -56,21 +56,23 @@ public class SchemaProcessor extends AbstractProcessor {
       }
 
       SchemaElement schemaElement = SchemaElement.get(context, element);
-      if (schemaElement.tableElement.declaredType != null) {
+      if (schemaElement == null) continue;
+      if (schemaElement.serializerElement.declaredType != null) {
         try {
-          tableCodeGenerator.generate(schemaElement)
+          serializerCodeGenerator.generate(schemaElement)
               .writeTo(processingEnv.getFiler());
-              // .writeTo(System.out);
         } catch (Throwable t) {
           context.error(ExceptionUtils.getRootCauseMessage(t));
           t.printStackTrace(System.err);
         }
       }
 
-      if (schemaElement.serializerElement.declaredType != null) {
+      // Depends on serializerElement to generate Serializer impl
+      if (schemaElement.tableElement.declaredType != null) {
         try {
-          serializerCodeGenerator.generate(schemaElement)
+          tableCodeGenerator.generate(schemaElement)
               .writeTo(processingEnv.getFiler());
+              // .writeTo(System.out);
         } catch (Throwable t) {
           context.error(ExceptionUtils.getRootCauseMessage(t));
           t.printStackTrace(System.err);
diff --git a/table/annotation-processor/src/main/java/com/riiablo/table/annotation/SerializerCodeGenerator.java b/table/annotation-processor/src/main/java/com/riiablo/table/annotation/SerializerCodeGenerator.java
index 6ae5ddae..e99c7f0f 100644
--- a/table/annotation-processor/src/main/java/com/riiablo/table/annotation/SerializerCodeGenerator.java
+++ b/table/annotation-processor/src/main/java/com/riiablo/table/annotation/SerializerCodeGenerator.java
@@ -14,6 +14,7 @@ import java.util.Arrays;
 import java.util.Objects;
 import javax.lang.model.element.Modifier;
 import javax.lang.model.element.Name;
+import org.apache.commons.lang3.Validate;
 
 import static com.riiablo.table.annotation.Constants.STRING;
 
@@ -23,8 +24,13 @@ class SerializerCodeGenerator extends CodeGenerator {
   }
 
   @Override
-  String formatName(SchemaElement schemaElement) {
-    return ClassName.get(schemaElement.element).simpleName() + Serializer.class.getSimpleName();
+  ClassName formatName(String packageName, SchemaElement schemaElement) {
+    Validate.validState(schemaElement.serializerClassName == null,
+        "schemaElement.serializerClassName already set to " + schemaElement.serializerClassName);
+    return schemaElement.serializerClassName
+        = ClassName.get(
+            packageName,
+            schemaElement.element.getSimpleName() + Serializer.class.getSimpleName());
   }
 
   @Override
diff --git a/table/annotation-processor/src/main/java/com/riiablo/table/annotation/TableCodeGenerator.java b/table/annotation-processor/src/main/java/com/riiablo/table/annotation/TableCodeGenerator.java
index acba6bae..e4c1de19 100644
--- a/table/annotation-processor/src/main/java/com/riiablo/table/annotation/TableCodeGenerator.java
+++ b/table/annotation-processor/src/main/java/com/riiablo/table/annotation/TableCodeGenerator.java
@@ -12,8 +12,10 @@ class TableCodeGenerator extends CodeGenerator {
   }
 
   @Override
-  String formatName(SchemaElement schemaElement) {
-    return ClassName.get(schemaElement.element).simpleName() + Table.class.getSimpleName();
+  ClassName formatName(String packageName, SchemaElement schemaElement) {
+    return ClassName.get(
+            packageName,
+            schemaElement.element.getSimpleName() + Table.class.getSimpleName());
   }
 
   @Override
@@ -23,6 +25,7 @@ class TableCodeGenerator extends CodeGenerator {
         .superclass(schemaElement.tableElement.declaredType)
         .addMethod(constructor(schemaElement))
         .addMethod(newRecord(schemaElement))
+        .addMethod(newSerializer(schemaElement))
         .addMethod(offset(schemaElement))
         .addMethod(indexed(schemaElement))
         .addMethod(primaryKey(schemaElement))
@@ -49,6 +52,18 @@ class TableCodeGenerator extends CodeGenerator {
         .build();
   }
 
+  MethodSpec newSerializer(SchemaElement schemaElement) {
+    TableElement tableElement = schemaElement.tableElement;
+    SerializerElement serializerElement = schemaElement.serializerElement;
+    return MethodSpec
+        .overriding(
+            tableElement.getMethod("newSerializer"),
+            tableElement.declaredType,
+            context.typeUtils)
+        .addStatement("return new $T()", schemaElement.serializerClassName)
+        .build();
+  }
+
   MethodSpec offset(SchemaElement schemaElement) {
     Schema config = schemaElement.annotation;
     TableElement tableElement = schemaElement.tableElement;
diff --git a/table/core/src/main/java/com/riiablo/table/Table.java b/table/core/src/main/java/com/riiablo/table/Table.java
index ae01d5f2..acfd6371 100644
--- a/table/core/src/main/java/com/riiablo/table/Table.java
+++ b/table/core/src/main/java/com/riiablo/table/Table.java
@@ -36,6 +36,7 @@ public abstract class Table<R> implements Iterable<R> {
   }
 
   protected abstract R newRecord();
+  protected abstract Serializer<R> newSerializer();
 
   public Class<R> recordClass() {
     throw new UnsupportedOperationException();
diff --git a/table/integration/src/main/java/com/riiablo/table/schema/MonStatsTableImpl.java b/table/integration/src/main/java/com/riiablo/table/schema/MonStatsTableImpl.java
index 3e0e38e3..3a24100d 100644
--- a/table/integration/src/main/java/com/riiablo/table/schema/MonStatsTableImpl.java
+++ b/table/integration/src/main/java/com/riiablo/table/schema/MonStatsTableImpl.java
@@ -1,5 +1,6 @@
 package com.riiablo.table.schema;
 
+import com.riiablo.table.Serializer;
 import com.riiablo.table.Table;
 
 public abstract class MonStatsTableImpl extends Table<MonStats> {
@@ -11,4 +12,9 @@ public abstract class MonStatsTableImpl extends Table<MonStats> {
   protected MonStats newRecord() {
     return new MonStats();
   }
+
+  @Override
+  protected Serializer<MonStats> newSerializer() {
+    return new MonStatsSerializerImpl();
+  }
 }