diff --git a/.fixtures.yml b/.fixtures.yml
index 7a72d436..d289934c 100644
--- a/.fixtures.yml
+++ b/.fixtures.yml
@@ -3,7 +3,5 @@ fixtures:
     apt: puppetlabs/apt
     elastic_stack: elastic/elastic_stack
     stdlib: puppetlabs/stdlib
+    concat: puppetlabs/concat
     zypprepo: darin/zypprepo
-
-symlinks:
-    logstash: #{source_dir}
diff --git a/manifests/init.pp b/manifests/init.pp
index c63a6068..62733ffe 100644
--- a/manifests/init.pp
+++ b/manifests/init.pp
@@ -151,7 +151,7 @@
   $settings          = {},
   $startup_options   = {},
   $jvm_options       = [],
-  Array $pipelines   = [],
+  Variant[Boolean,Array[Hash,1]] $pipelines = false,
   Boolean $manage_repo   = true,
 )
 {
diff --git a/manifests/pipeline.pp b/manifests/pipeline.pp
new file mode 100644
index 00000000..7ba68160
--- /dev/null
+++ b/manifests/pipeline.pp
@@ -0,0 +1,41 @@
+define logstash::pipeline
+(
+  Hash $config                     = {},
+  Optional[String] $content        = undef,
+  Optional[String] $source         = undef,
+  Optional[Stdlib::Unixpath] $path = undef,
+  Optional[String] $id             = undef,
+)
+{
+  unless $logstash::pipelines { fail('You must set base class `logstash`\'s `pipeline` parameter to `true` to use this defined type.') }
+  if $content and $source { fail('You can\'t specify both `content` and `source`') }
+
+  # overwrite keys in $config with values from $path and $id (if they're set).
+  $real_config = $config + {
+    'pipeline.id' => $id,
+    'path.config' => $path,
+    }.filter |$key, $val| { $val =~ NotUndef }
+
+  unless 'pipeline.id' in $real_config { fail("logstash::pipeline ${title} is missing a pipeline id") }
+
+  $yaml = [$real_config].to_yaml
+  $yaml_without_header = join($yaml.split("\n")[1,-1],"\n")
+
+  concat::fragment { "logstash pipeline ${title}":
+    target  => '/etc/logstash/pipelines.yml',
+    content => "${yaml_without_header}\n",
+  }
+
+  if $content or $source {
+    if 'path.config' in $real_config {
+      $real_path = $real_config['path.config']
+    } else {
+      fail('To use logstash::pipeline with `content` or `source`, the `config` hash must contain a `path.config` key or you must specify `path`')
+    }
+    logstash::configfile { "Config file for pipeline.id ${real_config['pipeline.id']}":
+      content => $content,
+      source  => $source,
+      path    => $real_path,
+    }
+  }
+}
diff --git a/manifests/service.pp b/manifests/service.pp
index d3b15b6a..2d85afc0 100644
--- a/manifests/service.pp
+++ b/manifests/service.pp
@@ -93,14 +93,29 @@
     # ..and pipelines.yml, if the user provided such. If they didn't, zero out
     # the file, which will default Logstash to traditional single-pipeline
     # behaviour.
-    if(empty($pipelines)) {
-      file {'/etc/logstash/pipelines.yml':
-        content => '',
+    if $pipelines {
+      # Either a non empty array of hashes, or true.
+      concat { '/etc/logstash/pipelines.yml':
+        ensure => present,
       }
-    }
-    else {
+      concat_fragment { 'pipelines.yml header':
+        content => "---\n",
+        target  => '/etc/logstash/pipelines.yml',
+        order   => 1,
+      }
+
+      if $pipelines =~ Array {
+        $pipelines.each |Hash $pipeline| {
+          $unique_resource_id = digest(String($pipeline))
+          logstash::pipeline { $unique_resource_id:
+            config => $pipeline,
+          }
+        }
+      }
+    } else {
+      # Traditional single-pipeline behaviour.
       file {'/etc/logstash/pipelines.yml':
-        content => template('logstash/pipelines.yml.erb'),
+        content => '',
       }
     }
 
diff --git a/metadata.json b/metadata.json
index 693c9257..015373ad 100644
--- a/metadata.json
+++ b/metadata.json
@@ -12,6 +12,10 @@
       "name": "puppetlabs/stdlib",
       "version_requirement": ">=3.2.0 <6.0.0"
     },
+    {
+      "name": "puppetlabs/concat",
+      "version_requirement": ">=4.1.0 <6.0.0"
+    },
     {
       "name": "elastic/elastic_stack",
       "version_requirement": ">=6.0.0 <7.0.0"
diff --git a/templates/logstash.yml.erb b/templates/logstash.yml.erb
index 5bafdaf3..bc3a4d09 100644
--- a/templates/logstash.yml.erb
+++ b/templates/logstash.yml.erb
@@ -4,7 +4,7 @@
 <%# removing the 'path.config' setting. -%>
 <%# -%>
 <%# REF: https://github.com/elastic/logstash/issues/8420 -%>
-<% @settings.delete('path.config') unless @pipelines.empty? -%>
+<% @settings.delete('path.config') if @pipelines -%>
 <%# -%>
 <%# Similiarly, when using centralized pipeline management, path.config -%>
 <%# is an invalid setting, and should be removed. -%>