diff --git a/lib/logstash/filters/mutate.rb b/lib/logstash/filters/mutate.rb
index a051420..cd12870 100644
--- a/lib/logstash/filters/mutate.rb
+++ b/lib/logstash/filters/mutate.rb
@@ -162,6 +162,22 @@ class LogStash::Filters::Mutate < LogStash::Filters::Base
   #     }
   config :merge, :validate => :hash
 
+  # Move all properties of a sub-structure of the event to the `target` field (which is the root level if not specified).
+  # By default, all the other properties of the target are kept, but it is also possible to erase them by setting `empty_target` to `true`
+  #
+  # Example:
+  # [source,ruby]
+  #     filter {
+  #       mutate {
+  #          move => {
+  #             "field" => "moved_field"
+  #             "target" => "target_field"
+  #             "empty_target" => true
+  #          }
+  #       }
+  #     }
+  config :move, :validate => :hash
+
   TRUE_REGEX = (/^(true|t|yes|y|1)$/i).freeze
   FALSE_REGEX = (/^(false|f|no|n|0)$/i).freeze
   CONVERT_PREFIX = "convert_".freeze
@@ -212,6 +228,7 @@ def filter(event)
     split(event) if @split
     join(event) if @join
     merge(event) if @merge
+    move(event) if @move
 
     filter_matched(event)
   end
@@ -427,4 +444,56 @@ def merge(event)
       end
     end
   end
+
+  def move(event)
+    if @move['field'].nil?
+      raise LogStash::ConfigurationError, I18n.t(
+          "logstash.agent.configuration.invalid_plugin_register",
+          :plugin => "filter",
+          :type => "mutate",
+          :error => "No field to move has been specified"
+      )
+    end
+
+    field = event.sprintf(@move['field'])
+    if event.get(field).nil?
+      @logger.warn("No field available in event", :field => field)
+      return
+    end
+
+    unless event.get(field).is_a?(Hash)
+      @logger.warn("Field to move must be a Hash", :field => field, :type => event.get(field).class)
+      return
+    end
+
+    # delete all target fields first?
+    if @move['empty_target'] and convert_boolean(@move['empty_target'])
+      # empty the root?
+      if @move['target'].nil?
+        event.to_hash.each do |k, v|
+          event.remove(k) unless k == field
+        end
+      else
+        # empty the target?
+        event.set(@move['target'], {})
+      end
+    else
+      # make sure that the target is a Hash and not a string, etc
+      unless @move['target'].nil?
+        unless event.get(@move['target']).is_a?(Hash)
+          event.set(@move['target'], {})
+        end
+      end
+    end
+
+    # move sub-fields to target level
+    event.get(field).each do |k, v|
+      target_key = k if @move['target'].nil?
+      target_key = "#{@move['target']}[#{k}]" unless @move['target'].nil?
+      event.set(target_key, v)
+    end
+
+    # delete moved sub-field
+    event.remove(field)
+  end
 end
diff --git a/spec/filters/mutate_spec.rb b/spec/filters/mutate_spec.rb
index 470034b..ae552a8 100644
--- a/spec/filters/mutate_spec.rb
+++ b/spec/filters/mutate_spec.rb
@@ -187,7 +187,7 @@ def pattern_path(path)
       config <<-CONFIG
         filter {
           mutate {
-            convert => [ "message", "int"] //should be integer
+            convert => [ "message", "int"] #should be integer
           }
         }
       CONFIG
@@ -638,4 +638,164 @@ def pattern_path(path)
     end
   end
 
+  describe "no sub-field to move" do
+    config '
+      filter {
+        mutate {
+          move => {
+          }
+        }
+      }'
+
+    sample("foo" => "bar") do
+      expect {subject}.to raise_error LogStash::ConfigurationError
+    end
+  end
+
+  describe "move empty sub-field at root level" do
+    config '
+      filter {
+        mutate {
+          move => {
+            field => "sub"
+          }
+        }
+      }'
+
+    sample("foo" => "bar") do
+      expect(subject.get("foo")).to eq "bar"
+    end
+  end
+
+  describe "move non-Hash sub-field at root level" do
+    config '
+      filter {
+        mutate {
+          move => {
+            field => "sub"
+          }
+        }
+      }'
+
+    sample("foo" => "bar", "sub" => "123") do
+      expect(subject.get("foo")).to eq "bar"
+      expect(subject.get("sub")).to eq "123"
+    end
+  end
+
+  describe "move sub-fields at root level" do
+    config '
+      filter {
+        mutate {
+          move => {
+            field => "sub"
+          }
+        }
+      }'
+
+    sample("foo" => "bar", "sub" => { "field1" => "value1", "field2" => "value2"}) do
+      expect(subject.get("foo")).to eq "bar"
+      expect(subject.get("field1")).to eq "value1"
+      expect(subject.get("field2")).to eq "value2"
+      expect(subject.get("sub")).to eq nil
+    end
+  end
+
+  describe "move sub-fields at root level and erase root fields" do
+    config '
+      filter {
+        mutate {
+          move => {
+            field => "sub"
+            empty_target => true
+          }
+        }
+      }'
+
+    sample("foo" => "bar", "sub" => { "field1" => "value1", "field2" => "value2"}) do
+      expect(subject.get("foo")).to eq nil
+      expect(subject.get("field1")).to eq "value1"
+      expect(subject.get("field2")).to eq "value2"
+      expect(subject.get("sub")).to eq nil
+    end
+  end
+
+  describe "move sub-fields to non-existing target" do
+    config '
+      filter {
+        mutate {
+          move => {
+            field => "sub"
+            target => "target"
+          }
+        }
+      }'
+
+    sample("foo" => "bar", "sub" => { "field1" => "value1", "field2" => "value2"}) do
+      expect(subject.get("foo")).to eq "bar"
+      expect(subject.get("[target][field1]")).to eq "value1"
+      expect(subject.get("[target][field2]")).to eq "value2"
+      expect(subject.get("sub")).to eq nil
+    end
+  end
+
+  describe "move sub-fields to existing hash target" do
+    config '
+      filter {
+        mutate {
+          move => {
+            field => "sub"
+            target => "target"
+          }
+        }
+      }'
+
+    sample("foo" => "bar", "sub" => { "field1" => "value1", "field2" => "value2"}, "target" => { "field3" => "value3" }) do
+      expect(subject.get("foo")).to eq "bar"
+      expect(subject.get("[target][field1]")).to eq "value1"
+      expect(subject.get("[target][field2]")).to eq "value2"
+      expect(subject.get("[target][field3]")).to eq "value3"
+      expect(subject.get("sub")).to eq nil
+    end
+  end
+
+  describe "move sub-fields to existing hash target and erase target fields" do
+    config '
+      filter {
+        mutate {
+          move => {
+            field => "sub"
+            target => "target"
+            empty_target => true
+          }
+        }
+      }'
+
+    sample("foo" => "bar", "sub" => { "field1" => "value1", "field2" => "value2"}, "target" => { "field3" => "value3" }) do
+      expect(subject.get("foo")).to eq "bar"
+      expect(subject.get("[target][field1]")).to eq "value1"
+      expect(subject.get("[target][field2]")).to eq "value2"
+      expect(subject.get("[target][field3]")).to eq nil
+      expect(subject.get("sub")).to eq nil
+    end
+  end
+
+  describe "move sub-fields to existing non-hash target" do
+    config '
+      filter {
+        mutate {
+          move => {
+            field => "sub"
+            target => "target"
+          }
+        }
+      }'
+
+    sample("foo" => "bar", "sub" => { "field1" => "value1", "field2" => "value2"}, "target" => "123" ) do
+      expect(subject.get("foo")).to eq "bar"
+      expect(subject.get("[target][field1]")).to eq "value1"
+      expect(subject.get("[target][field2]")).to eq "value2"
+      expect(subject.get("sub")).to eq nil
+    end
+  end
 end