# alasdairb.com > The blog of Al Brown. Posts about data, developer experience, and tooling. Full text of every published post, newest first. # is-ai-agent for Rust: detect if your CLI is invoked by an agent Source: https://alasdairb.com/posts/is-ai-agent-rust-crate Published: 2026-05-04 I published a new Rust crate [is-ai-agent](https://crates.io/crates/is-ai-agent). It lets you detect when your CLI is being called an AI agent, by detecting standard/common environment variables that various agents set. It's a very simple API: ```rust use is_ai_agent::{detect, is_ai_agent}; if is_ai_agent() { // emit structured/agent-friendly output } if let Some(agent) = detect() { eprintln!("running under {}", agent.name); } ``` It's not perfect, as there's no standard signal for agents to identify themselves yet. There is an ongoing [discussion for creating a standard `AGENT` variable](https://github.com/agentsmd/agents.md/issues/136), but it seems that only a few of the smaller open source agents have adopted it so far. For now, this is the environment variable and agent coverage: | Variable | Agent | |---|---| | `CLAUDECODE`, `CLAUDE_CODE` | Claude Code | | `CURSOR_TRACE_ID` | Cursor (editor) | | `CURSOR_AGENT`, `CURSOR_EXTENSION_HOST_ROLE=agent-exec` | Cursor CLI | | `GEMINI_CLI` | Gemini CLI | | `CODEX_SANDBOX`, `CODEX_CI`, `CODEX_THREAD_ID` | OpenAI Codex | | `ANTIGRAVITY_AGENT` | Antigravity | | `AUGMENT_AGENT` | Augment | | `CLINE_ACTIVE` | Cline | | `OPENCODE_CLIENT` | OpenCode | | `TRAE_AI_SHELL_ID` | TRAE AI | | `GOOSE_TERMINAL` | Goose | | `REPL_ID` | Replit | | `COPILOT_MODEL`, `COPILOT_ALLOW_ALL`, `COPILOT_GITHUB_TOKEN` | GitHub Copilot | I needed this for [clickhousectl](https://github.com/ClickHouse/clickhousectl), so I'll be keeping the rules up to date. There's a [similar lib for TypeScript with @vercel/detect-agent](https://www.npmjs.com/package/@vercel/detect-agent). --- # Agent Experience: Building a CLI for ClickHouse Source: https://alasdairb.com/posts/agent-experience-building-a-cli Published: 2026-02-16 > **Update:** Since writing this post, `chv` has evolved into [clickhousectl](https://github.com/ClickHouse/clickhousectl), the official CLI for ClickHouse. The concepts and lessons below still apply, but the tool is now maintained by ClickHouse and available as `clickhousectl`. Developer experience (DX) is now old and boring, and it's all about **agent experience** (AX). As it turns out, they're practically the same thing. Developers write code by slapping buttons with the meat sausages attached to their arms. LLMs write code by generating it like magic. A good experience allows them do what they need to do without thinking too hard, or needing to lower themselves to using a UI. The platforms with the best AX today are, unsurprisingly, the platforms that had the best DX yesterday. Like Supabase. They've had a great SDK and CLI that meat-based users have enjoyed using for years. And now LLMs love using them too. ## Why am I thinking about this? I build with ClickHouse a lot, and while I've pretty much delegated all of the actual code-writing to Claude, I still find myself manually managing the local and cloud infra. I don't need the repeatability of IaC, I just want to spin new stuff up with less effort. ClickHouse has various language clients that let you integrate it with your app, and LLMs are great with those, but they don't help you from the infrastructure side. If you want to build an app using ClickHouse with an LLM, you probably need to be comfortable installing and setting it up yourself. And when you want to move to ClickHouse Cloud, you'll need to do that yourself, too. So I wanted to see what it takes to close the gap. How can I touch nothing except the prompt? ## Designing a CLI To me, the first logical step was a CLI. When I build apps, its usually with JavaScript or Python, and LLMs do great with `uv` or `pnpm` to manage Python/Node environments and dependencies. Maybe I could treat having ClickHouse like a Node environment; what if I had a `pnpm env` style CLI to install and manage my local ClickHouse? So I started on [chv](https://github.com/ClickHouse/clickhousectl). > **Disclaimer: `chv` is a personal project of mine, and not in any way supported by ClickHouse Inc. You're welcome to try it, but know that it's just an experiment, and the code is 95% written by Claude and hasn't been reviewed by a competent human.** The initial set of commands pretty much replicated what I could do with `uv` and `pnpm`, with the subcommands: 1. init 2. install 3. use 4. list With these, I can easily find and pull down versions of ClickHouse and bootstrap my working dir. Now, agents have web tools, and they *could* go to GitHub, pull down the releases, parse it to find the version it wants and then form a `curl` commands. But this is far more expensive in context & tokens than `chv use stable`. I also need ways to run a ClickHouse server and iteract with it. LLMs are great at `pnpm run dev` so...why not `chv run`? 5. run Now I can do `chv run server` to start a ClickHouse server, and `chv run client` to run the client to connect and interact with my sever. With those 5 commands, I can actually get pretty far. But...how to get an LLM to understand and use it? ## Self-discovery and Agent Skills LLMs haven't been trained on my CLI. They don't even know it exists. So how can I get them to use it? An LLM not knowing about my CLI is a simply a problem of it not existing in the world. And I can solve that by releasing it and talking about it. There's not much I can do to make LLMs know it exists while it's a bag of bits in my home dir. So, I focused on getting them to understand using it. LLMs know how to use CLIs. And what CLI doesn't have `--help`? LLMs know this convention, and you can see them use it all the time. They are able to use `--help` to self-discover how to use a CLI with pretty decent results. So, the first step was to make sure I had good quality help text. That helped noticably, but I still found the agent making mistakes. They could work out how to use commands, but not always which commands to use together. And they would often resort to running `--help` on every single combination of subcommand. So I added a paragraph of text to every command's help output, aimed at agents. Its heavier than just listing commands with short descriptions, but by encoding common flows into this, the LLM stopped listing *everything* and focused on commands it thought it needed. Not always...but enough that I considered it a win. ``` CONTEXT FOR AGENTS: chv is a CLI to work with local ClickHouse and ClickHouse Cloud. Two main workflows: 1. Local: Install and interact with versions of ClickHouse to develop locally. 2. Cloud: Manage ClickHouse Cloud infrastructure and push local work to cloud. You can install the ClickHouse Agent Skills for best practices on using ClikHouse: `npx skills add clickhouse/agent-skills` Typical local workflow: `chv install stable && chv use stable && chv run server`. Use `chv --help` to get more context for specific commands. ``` ### Agent Skills I also created some Agent Skills that encode even deeper knowledge of using the commands. I forked the [official ClickHouse Agent Skills](https://github.com/ClickHouse/agent-skills), which already has skills to help LLMs write better SQL and schemas, and created two new skills with various guides on using the CLI. And called out in the CLI help that it should install and use the skills. This also worked pretty well. It did install the skills, and it did use them...sometimes. When it used them, the results were great. But I found that, even using the latest Opus 4.6, skill invocation is frustratingly inconsistent. My feeling is that building Agent Skills for your CLI is worthwhile, particularly if you're developing something new and LLMs haven't been trained on piles of docs and examples. And I'm hopeful that skill utilisation is going to continue to improve in the models. ## Going to Cloud With all that done, the CLI was working pretty well. I could go into Claude Code, and say ``` Build me a competitor to Google Analytics. Use ClickHouse as the database, you have the ClickHouse CLI `chv` installed. ``` and it could one shot it somewhat-reliably. Install ClickHouse, start the server, bootstrap a Nextjs app, build a Chartjs dashboard, add the ClickHouse JS client, create my tables, populate fake data, write the queries and wire it all up. I didn't have to do anything. But, my friends said they couldn't reach it when I shared the app with them - try it out - [`http://localhost:3000`](http://localhost:3000). So I need to be able to deploy it. Agents can already work with Vercel to deploy the Nextjs part, but not with ClickHouse Cloud to take my database to prod. The sixth subcommand: 6. cloud ClickHouse Cloud already has a REST API and an OpenAPI spec, but LLMs aren't good at using complex REST APIs. When they have to take in the full JSON OpenAPI spec, and form raw cURLs with big JSON payload, they get it wrong. And, most of the time, it seems like they pretend they don't exist - I assume that is because most written examples are using CLIs! When I started on the `cloud` subcommand, it was the day that Anthropic released Opus 4.6 (I had been using 4.5). So that was its first test: I gave it the full OpenAPI spec, and told it to write a wrapper for it under the `cloud` prefix. I'd say it got about 80% there. It missed various properties, didn't read descriptions that called out deprecations, etc. so I had to fix it up, but it was certainly close enough to save me some time. This command let me operate all the usual bits of ClickHouse Cloud I need to get to prod: find my org, create a service, get my connection details. I could now complete the loop, and after my agent was done building my app locally, I could ask it to push to cloud. The first time it successfully spun up all my cloud infra, pushed my work up, and toggled the app over was pretty enjoyable. An interesting problem that I don't think is solved all that well for agents is the boring admin stuff - auth, users management, and billing. I've seen some folks working on these areas, and I'm assuming we'll eventually get some kind of MCP-style spec for agentic auth. I don't think its wise to roll a bespoke experience yourself, and frankly, a waste of time that you could use to do something more valuable. ## Playing with OpenClaw I had resisted playing with OpenClaw, but a colleague was messing with it and I couldn't resist. I had just been using Claude Code, and I felt like a cave man compared to him chatting with his via WhatsApp. So, in a few hours, I had a VPS on Hetzner running OpenClaw, and revived an old phone to keep things separate. Could I really build an app and push it to cloud, with just my phone, from the sofa? Yes. Not first time, of course. I built `chv` on a Mac, and never tested it on Linux. My VPS was using Fedora, and it failed to install ClickHouse using `chv`, went down the non-CLI route and got itself stuck. With `chv` 0.1.8 supporting Linux, I returned to my sofa. And it worked first time. I asked my agent to build a security manager for OpenClaw. It installed ClickHouse, ran the sever, set up my schemas, ingested SSH server logs, built a Node app, connected the two, designed security rules for hardening OpenClaw (mostly just hardening Linux)...and pushed it all to cloud when it was done. The app isn't particularly novel, we've been able to vibe code this stuff for at least a year. But the CLI workflow it was using to develop against ClickHouse did not exist until 2 hours prior. ## Lessons I thought I'd have to do a lot more. I didn't relish the idea of building an SDK, but in the end, I didn't have to. I suspect an SDK would allow some really cool stuff - what if most of this infra work was just inferred from our code? - but a super simple CLI with 6 subcommands enabled agents to take another job off of my hands. This is already too long of a post, and I can't be bothered to write "The 10 Lessons of Agent Experience". So I'll leave you with 3 things that I think are inarguably true: 1. LLMs consume & generate text. Focus on text. CLIs are back, baby! 2. LLMs like patterns. Be boring and predictable. 3. LLMs can understand tools, but don't always know how to use them to accomplish a goal. It *might* be cheaper to give this guidance upfront, rather than let an LLM go down its own rabbit hole. --- # How to create a Python Processor in Apache NiFi 2 Source: https://alasdairb.com/posts/building-a-nifi-processor-in-python Published: 2024-09-02 (The start of this post is a bit of a "What this bolognese recipe means to my family"-style rant. If you just want the steps, [skip ahead](#from-0-to-hello-world-steps)) I love NiFi, it's a great tool, and NiFi 2 added support for Python Processors. This was pretty exciting 'cause I hate working with Java (the language itself is...fine, but the ecosystem and developer experience are terrible). I've been using NiFi on-and-off since 2018-ish, and worked at Cloudera (the primary corporate sponsor of NiFi) for a few years doing a lot of NiFi work. As excited as I was, and confident in my NiFi knowledge, I struggled to get to a 'Hello World' moment with Python Processors. While I might be a bit out of date, I found the documentation practically useless from a 'getting started' perspective, and NiFi's 'I just want to play' experience is, and has been for some time, just down right awful. Part of this is due to the secure-by-default decision, made somewhere around version 1.14.0. It somewhat made sense at the time (I was working at Cloudera when this decision was made); NiFi came from the security world and found most of its early adoption in more secure, enterprise environments. Forcing these enterprises to follow sane security best practices stopped people doing stupid things and getting into trouble later on... ...but, setting up security on NiFi isn't trivial, particularly if you're not close to the "Java Way"...and if you're excited by Python Processors, it's probably *because* you're not close to the "Java Way". It is, frankly, a truly shitty first impression. It was already hard when 1.14.0 was relevant and docs/examples were up-to-date, but now that most of that content is outdated and useless, it's even more of a confusing mess. It's quite sad because, once you get past it, NiFi is such an incredible tool, but expecting people to fight through this clusterfuck of a setup...it's just not going to happen, and it will kneecap adoption. This is my fourth attempt at getting a 'Hello World' example running, I fought and gave up 3 times before finally getting it working. I know it's open source, I know I could contribute. But this needs directional agreement, and I likely wouldn't be fluent enough in Java to contribute it anyway. So this is my contribution: a guide to help people like me, and a plea to the NiFi team (who are all amazing btw), please, please, please, please...improve the NiFi onboarding experience to not suck quite this much. ## From 0 to 'Hello World' steps This was done on an M1 Macbook Pro, using OrbStack to run a Linux VM (Ubuntu). ### Set up your machine I use [OrbStack](https://orbstack.dev/) to run VMs & containers on my Mac. For this guide, use an Ubuntu 24.04.1 VM. Name the VM `nifi`. If you used OrbStack, the machine's hostname is configured as `nifi` and this will create a local DNS name of `nifi.orb.local`. **This is important, because the TLS configuration of NiFi absolutely requires hostnames and will not work with IP addresses.** Access the terminal of the VM. [Install JDK 21](https://askubuntu.com/questions/1492571/install-openjdk-21) with: ```bash sudo apt install openjdk-21-jdk ``` And verify with: ```bash java -version ``` Now [configure your JAVA_HOME](https://askubuntu.com/questions/175514/how-to-set-java-home-for-java). Edit `/etc/environment` and add a line at the bottom (replace the path with your own, it'll be under `/usr/lib/jvm/`): ```bash JAVA_HOME="/usr/lib/jvm/java-21-openjdk-amd64" ``` Then run: ```bash source /etc/environment ``` Now we need to set up Python 3.11 (as this Ubuntu ships with 3.12, which isn't supported by NiFi yet): ```bash sudo apt install software-properties-common sudo apt update sudo add-apt-repository ppa:deadsnakes/ppa sudo apt update sudo apt install python3.11 python3.11 --version ``` Finally, install `wget` and `unzip`: ```bash sudo apt install wget unzip ``` ### Setting up NiFi 2 Download NiFi 2.0.0 M4 and unzip it: ```bash wget https://dlcdn.apache.org/nifi/2.0.0-M4/nifi-2.0.0-M4-bin.zip unzip nifi-2.0.0-M4-bin.zip ``` Edit the `nifi-2.0.0-M4/conf/nifi.properties` file. Change the `nifi.web.http.host` to `0.0.0.0`. Then update `nifi.python.command=python3` line to `nifi.python.command=python3.11` and remove the `#` from the start of the line. Save the file. Then download the **previous version** of NiFi Toolkit (1.27.0) and unzip it: ```bash wget https://dlcdn.apache.org/nifi/1.27.0/nifi-toolkit-1.27.0-bin.zip unzip nifi-toolkit-1.27.0-bin.zip ``` You need the old version because `tls-toolkit` was deprecated with NiFi 2.0 and no longer ships with it. This makes it even more of a pain in the ass to do the initial TLS setup. Luckily, the old version of `tls-toolkit` still works and generates valid certs & config, so we can use it for now. Run `tls-toolkit` like so (make sure to edit the `nifi.orb.local` if your FQDN is different): ```bash nifi-toolkit-1.27.0/bin/tls-toolkit.sh standalone -n 'nifi.orb.local' ``` This will create a folder called `nifi.orb.local` which contains 2 `.jks` files and a `nifi.properties` file. Move the two `jks` files to `nifi-2.0.0-M4/conf/`, for example: ```bash mv *.jks nifi-2.0.0-M4/conf/ ``` Next, open the generated `nifi.properties` file (the one that `tls-toolkit` created inside the `nifi.orb.local`), and search for this part: ``` nifi.security.autoreload.enabled=false nifi.security.autoreload.interval=10 secs nifi.security.keystore=./conf/keystore.jks nifi.security.keystoreType=jks nifi.security.keystorePasswd=JfvOBGE6i1j+FsHKzvvv4cUjoYEH7HnQBkSjo7SRSB4 nifi.security.keyPasswd=JfvOBGE6i1j+FsHKzvvv4cUjoYEH7HnQBkSjo7SRSB4 nifi.security.truststore=./conf/truststore.jks nifi.security.truststoreType=jks nifi.security.truststorePasswd=ckQ5e55245alswyQd0w/ZkuEO1lChTC6mbNv4S/lxio ``` Copy all of these lines. Now go open the real `nifi.properties` file in `nifi-2.0.0-M4/conf/nifi.properties`, find the same lines, delete them, and replace them with the new ones. ### We're almost there! On to Python & the NiFi UI Create the extensions dir: `nifi-2.0.0-M4/python/extensions`. I don't know why this isn't created by default, when it's configured by default, but there you go šŸ¤·ā€ā™‚ļø. Create a file for your `Hello World` extension: `nifi-2.0.0-M4/python/extensions/MyProcessor.py` Paste in the following contents to the file: ```python from nifiapi.flowfiletransform import FlowFileTransform, FlowFileTransformResult class WriteHelloWorld(FlowFileTransform): class Java: implements = ['org.apache.nifi.python.processor.FlowFileTransform'] class ProcessorDetails: version = '0.0.1-SNAPSHOT' def __init__(self, **kwargs): super().__init__(**kwargs) def transform(self, context, flowfile): # Import Python dependencies input = flowFile.getContentsAsBytes().decode() # Do something with the input output = input return FlowFileTransformResult( relationship = "success", contents = output, attributes = {"greeting", "hello"} ) ``` Give it a save. Now start NiFi with: ```bash nifi-2.0.0-M4/bin/nifi.sh start ``` On your host machine, open your browser and go to `https://nifi.orb.local:8443/nifi/`. You'll get a warning about the cert being self-signed, but you can ignore it and proceed. You should reach the login screen asking for a user & pass. You'll need to get these from the NiFi logs. Run: ```bash cat logs/nifi-app.log | grep Username cat logs/nifi-app.log | grep Password ``` You should see lines like these: ```bash Generated Username [ba640a89-1532-42f7-bbc4-51f194a75547] Generated Password [1qmQNTnkFLXHlJ9R7kAaRNBE0CvuPeJz] ``` Copy and paste these into the login screen. If all went well, you should now be into the NiFi canvas! Drag a new processor to the canvas, search for `MyProcess`, and you should see your `Hello World` processor ready to go! ## Oh how painful! Once you reach this point, the NiFi docs are actually in reasonable shape to work with the Python Processor API. Hopefully this was enough to get you passed the absolute nightmare that is the initial setup. Again, I love NiFi, and the team working on it are amazing people that I've had the privilege to work with...but I hope they find some time to revisit the onboarding experience cause it suuuuuuuuuuucks. If you didn't know, most of the NiFi team have left Cloudera and formed [Datavolo](https://datavolo.io/) around NiFi, which is pretty cool! --- # Conversations with 200+ people at Kafka Summit Source: https://alasdairb.com/posts/200-conversation-at-kafka-summit Published: 2024-04-28 I spent 2 days at Kafka Summit 2024 and spoke with over 200 people. They were loooong days. But itā€™s an amazing way to completely surround yourself with people who are actually doing the job, and thatā€™s such a fantastic opportunity to learn. Thereā€™s plenty of posts out there that summarize the talks, recap the keynote and hype the vendor releases, so Iā€™m not going to talk about any of that. This post is just going to cover the conversations that I had with practitioners that attended the event. Given that was over 200 conversations, this is going to be my personal distillation of what we spoke about. Of course, thereā€™s some bias here: itā€™s a Kafka event so itā€™s generally folks with streaming on their mind, I have my own biases that will influence the natural path conversations take, and I was there representing my company. I believe the topics are still useful with these biases in mind. The four main themes that I found myself discussing were: - [Operational Analytics](#operational-analytics) - [Event Driven Architectures](#event-driven-architectures) - [Unified ETL with Apache Flink](#unified-etl-with-apache-flink) - [Automation and Data as Code/Config (DaC)](#automation-and-data-as-codeconfig-dac) ## Operational Analytics I canā€™t count the amount of conversations I had around this one. So-called ā€œoperationalā€ systems tend to be the backends of a businessā€™ application, often described with a combination of terms like relational, transactional, CRUD, ACID or documents - think databases like Postgres or MongoDB. Generally, they need to support a lot of key-based single-row READs, as well as single-row DELETE and UPDATEs. The ā€œanalyticalā€ systems, on the other hand, are powering the companiesā€™ internal reporting and business intelligence. Here, the name of the game is generally ā€œCan I stick Tableau on it and do crazy JOINs over a few TBs of data?ā€. Itā€™s going to power some pretty crazy queries that will scan a huge amount of historical data to try and answer a question. However, many attendees are starting to see that operational systems, or the applications that use them, now want to utilize analytics. Most often, this is in the form of user-facing analytics, where the output of analytical queries can be served back to users within the application. This appears to be causing quite the headache, as there is typically a large divide between the ā€œoperationalā€ and ā€œanalyticalā€ teams or systems, and existing ā€œoperationalā€ systems cannot handle the analytical workload. Simply put, the operational team canā€™t just start running massive analytical queries over their Postgres database without tanking the service, and building a solution over the internal Snowflake is too slow & cumbersome. ā€œOperational analyticsā€ seemed to be the most popular term for this idea, which makes sense. ## Event Driven Architectures Event Driven Architectures (EDA) isnā€™t a new subject for a Kafka conference, Iā€™d say itā€™s probably one of the longest living themes (though definitely behind ā€œKafka is a pain to manageā€). In the past, Iā€™ve found that EDA is usually something that smaller, newer businesses are very keen on, but remains largely unexplored by larger enterprises. It felt a bit different this year, with many of these conversations occurring with engineers from global giants. The conversations were largely the same as they always were; folks are tired of pushing data into silos then building a bunch of glue around it to work out if they need to take action, and then more glue to actually take the action. The challenge here is that much of this ā€œglueā€ requires domain knowledge of the systems on either side: you need to know where the data lives and how its stored, write logic against that model to determine if action is required, and then translate this into whatever the destination system needs. Because of the inter-domain nature of this glue, itā€™s common that ownership of the glue itself is unclear. Itā€™s also typically quite brittle, as systems on either side may change, and there is often a lack of communication to notify other teams of changes that may break things. Instead, new data should be treated as events that represent something happening, be pushed onto a central bus, and downstream systems can subscribe to the stream(s) of events that they care about. This means that any downstream systems and team has a single, known and consistent integration point when needing access to data. This makes it much easier for each domain to self-service access to data, keep full ownership of their systems and make changes without breaking other consumers. Of course, you can also become ā€œevent drivenā€, so rather than polling every 5 minutes and working out all of the actions you should have taken, you can start to adopt an ā€œalways-onā€ pattern, where events trigger individual actions as they arrive. This is probably what jumps into most people minds when they think of the benefits or purpose of EDA...but what I took away from all of these conversations is that the structural benefits of decoupling event ingestion & distribution from the myriad of downstream systems is probably the more immediate benefit to a lot of teams. ## Unified ETL with Apache Flink Which brings us to Apache Flink, which was the hot topic of Confluent and many other vendors at Kafka Summit this year. Most people, even those already deep into it, will agree Flink is a pretty complex system. However, it brings some really nice things with it. It has been built to be ā€˜streaming firstā€™, making it a fantastic choice when working with streaming data. And while it excels at streaming, Flinkā€™s design is flexible and allows it to work with batch data as well, meaning you can reuse knowledge across both paradigms. On top, it has a growing ecosystem of input and output integrations, a SQL abstraction called ā€˜Flink SQLā€™ and has been tried and tested at a huge scale. (If you consider the EDA pattern described above, you might see how Flink positions quite nicely between a central streaming bus and anything else to the right hand side of it.) The core purpose of Flink was to be a stream processing engine, but as is often the case, it seems the market is finding that thereā€™s a different, and perhaps more widely applicable, fit for it. Its flexibility in working with both streaming & batch makes it attractive in the oft varied environments where ETL tooling becomes painful. The ability to write code is attractive to handle complex scenarios, while the comprehensive SQL abstraction makes it adoptable by teams who donā€™t have the resources to be effective with, or simply donā€™t need, that level of control. And the huge range of integrations make it a no brainer. Perhaps what weā€™ll see is that teams initially adopt Flink to solve ETL, and then ramp into stream processing use cases in the future. ## Automation and Data as Code/Config (DaC) ā€œDo you have a Terraform provider?ā€ When I'm repping an event for Tinybird, I get asked this question a lot. People want to know if they can deploy & build with the platform using Terraform. To be honest, Iā€™m still surprised at the level of adoption Terraform has seen in data teams, but it seems quite well entrenched now. Though, I do wonder if that will change with IBM's acquisition of HashiCorp. Do people really want to be beholden to Ol' Big Blue? Iā€™ve seen this idea of ā€˜Data as Codeā€™ (or ā€˜as Configā€™) deployed to spectacular results. Itā€™s a pretty simple concept where every part of a Data Platform is defined in files - some folks call it Code, and others Config, generally itā€™s the same idea - including the platform itself, schemas, integrations, queries, jobs and all artifacts of actual use cases. Perhaps the most striking benefit is the impact it has on how people work. If you are in a data team, or working with one, youā€™ll probably be familiar with the pains around collaboration - who performs what work, who owns it, who reviews it, who supports it, etc. Some teams have gone for clear centralization, where the data team is the gatekeeper to everything "data", while others have gone the ā€˜data meshā€™ route, and federated as much as possible into domain teams. Both approaches have their benefits, but in reality, neither approach has perfectly solved every pain. By making this (mostly technical) change, both the centralized and decentralized ways of working are improved, but it opens a nice path to **ā€œcontrolled decentralizationā€**. There are significant benefits to a data team being able to make well-informed decisions about data infrastructure, and centralizing knowledge and experience makes it easier to support and appropriately resource work. But weā€™re all aware of the discussion around ā€˜domain expertiseā€™ and the push for data teams to be ā€˜closer to the businessā€™. This makes sense, but weā€™ve been saying it for over a decade and it isnā€™t happening. I think itā€™s pretty unrealistic to expect that data teams will become self-sufficient in business domain knowledge. Businesses are too different, even within a single industry, let alone across industries (and most data engineers donā€™t stay in the same industry their whole career). Instead, this DaC model allows us to centralize knowledge while federating responsibilities. The responsibility for defining and owning the data platform is given to the data team, who are best placed for it. However, that knowledge is centralized in a repository that is open to all. Similarly, the specifics of building use cases are federated to the teams who understand them best, but they also push that knowledge to the central repository. The repository now becomes a place for those two teams to collaborate. The business teams can work at their own pace to build functionality, while the data team can put guardrails in place that allow for most ā€˜overheadā€™ work (e.g. deploying) to be automated. --- # Keeping up with WarpStream Source: https://alasdairb.com/posts/keeping-up-with-warpstream Published: 2024-04-26 Thereā€™s been quite a few ā€œKafka alternativesā€ come onto the market, particularly in the past ~5 years. It makes sense; Kafka has had immense adoption, but itā€™s fair to say that ever since Day 1, itā€™s never been particularly easy. The ecosystem around Kafka is vast, and a huge percentage of that is just tooling that attempts to make it easier to manage (let alone be productive with). That is a pretty obvious opportunity. To try and make it easier, thereā€™s all kinds of managed Kafka platforms: Confluent, Aiven, Upstash, DoubleCloud, AWS MSK, Azure Event Hubs. Thereā€™s also platforms that serve the same (or similar) purpose that arenā€™t based on Kafka: Apache Pulsar, StreamNative, Memphis.dev, JetStream, Fluvio, Redis Queues. Not too long ago, Redpanda came along and rewrote Kafka from the ground up. Rather than Java, it was C++. Instead of ZooKeeper, it was Raft. And many more architectural changes that made it a very attractive modern replacement for Apache Kafka. Itā€™s a great example of saying ā€œThis was great, but we could do it better today.ā€ But largely, all of these follow a very traditional platform architecture. Theyā€™re built expecting traditional computing primitives: servers, with CPUs and memory, and disks and sending data between machines on the networks. Duh, right? But a lot of todayā€™s computing is done on the cloud, where you donā€™t have to build with those same primitives. This is what I find particularly interesting about [WarpStream](https://www.warpstream.com/); itā€™s being built from the ground up expecting todayā€™s cloud primitives. Forget JBOD, data striping, partitioning, and tiered-storage - weā€™ve already got S3...and so on, challenging each decision on how to architect an event streaming platform for the cloud. I like that. If thereā€™s anything I can say about myself, itā€™s that I hate ā€˜traditionā€™. Challenging the assumptions of the past is always a good thing, itā€™s how we make progress. And as a data streaming nerd, it excites me to see that applied quite radically to an event streaming platform. --- # There is no Data Engineering roadmap Source: https://alasdairb.com/posts/there-is-no-data-engineering-roadmap Published: 2023-07-08 Between Reddit, twitter, LinkedIn and various Slack communities, I see multiple junior folk looking to break into Data Engineering and asking for advice. Every single day. Many ask for a "roadmap" or some kind of step by step lesson plan that will land them their dream job. I don't believe that such a roadmap exists. ## Newbies are welcome šŸ‘‹ I have seen some say "Data Engineering is not an entry level role" and this is nothing more than toxic gatekeeping. Data Engineering is no more, and no less, complex than any other software discipline. Every discipline is open to newbies. If you want to get into data, you can do it. You don't *need* to "graduate" into it from a different discipline. Moving sideways into Data Engineering is very common, not because it's necessary, but because Data Engineering is relatively new as a somewhat well-defined job category. Data Engineering teams haven't been commonplace for long, in fact, there's many industries still just starting to catch on. Many Data Analysts and Software Engineers already have at least *some* level of hands-on experience with data, so it makes total sense to use & develop those skills. This has happened with every single kind of engineering role in the history of engineering. But there's only so many people who want to make the switch, and you can't reallocate *everyone* from your other teams. So, Data Engineering absolutely needs entry-level engineers. People used to say "Software Engineering is not an entry level role". They don't anymore, because people know it is total rubbish. Everyone is welcome in data. ## All you need is ~~love~~ SQL ā¤ļø So, how does an entry-level engineer get started in Data Engineering? Firstly, go unfollow all those influencers on LinkedIn and Twitter. You don't need them, in fact, they are dangerous. They're not here to guide, help or teach you. They will take you down a path of failure so that you are more open to giving them your money for a quick win ([rant](#šŸŒ¶ļø-a-quick-rant-šŸŒ¶ļø)). With that out the way, understand that there is no roadmap. There is no single path, no clear linear progression of knowledge. No one can tell you that you absolutely must learn A, then B, then C and you're guaranteed to be a successful Data Engineer. The same applies to pretty much all engineering roles; front end, back end, embedded systems, networking, analytics. Whatever. In all of these cases, there are basics that *everyone* should get familiar with, and these are usually enough to get you your first gig. Remember, there is a fundamental difference between "What do I need to get my first job?" and "How do I progress my career?". For Data Engineering, there is only one skill that is absolutely, non-negotiably, the first thing you should learn to get started. **SQL.** Yep, SQL. It's not dead. It never will be. SQL is the cockroach of data and it's not going anywhere. People have tried to displace it, and they have all failed. ![](./images/there-is-no-data-engineering-roadmap-1.jpg) SQL is the only skill that every single Data Engineer uses every single day. No other skill or tool can claim the same. Python is common, some folks are using Scala, Snowflake is popular...but there's more data teams *not* using those tools than those who *are*. But not for SQL. You're an entry level engineer, you don't need to be an expert in SQL. You need to be able to solve problems. When you write some shit SQL, and you absolutely will, one of two things will happen: Thing 1, someone will tell you it's shit and you'll learn how to do it better. Thing 2, people will thank you for solving the problem and ask you to do something else. Win win. ## How should I learn SQL? If you're new to SQL and databases, you should know that "SQL" is very poorly standardised. You'll hear folks say "ANSI SQL" which many think is some kind of standard, but it's not really. Anyway, if that topic interests you, [read up on it](https://blog.ansi.org/ansi/sql-standard-iso-iec-9075-2023-ansi-x3-135). There is a common SQL base, but pretty much every single database in existence customises and extends SQL to do whatever it wants. This means that SQL you write for Postgres *might* work in MySQL, but don't be surprised if it doesn't. The same is true across Microsoft's SQL Server, Oracle Database, BigQuery, Redshift, Snowflake, ClickHouse and any other database you can think of. The different flavours of SQL used by these databases are called "dialects". This can make it daunting to get started, but it's no different than getting into Software Engineering. There's a million programming languages and most people start with one and then try a bunch of others. So, pick any database, don't worry about the dialect. The database is simply a vehicle for you to learn SQL. Stuck? Start with [Postgres](https://www.postgresql.org/). It's the world's favorite free, open source database. You can't go wrong starting with Postgres. Follow some simple tutorials; get it set up, load some data and start asking questions with SQL. Google around for "SQL challenges", there's loads. Some are better than others, just go through them all and challenge yourself. As your knowledge improves, look for harder problems and bigger data sets. When you're starting to feel confident - change database. Try solving the same problems with [MariaDB](https://mariadb.org/). Then try out [Google's BigQuery](https://cloud.google.com/bigquery) (there is a generous free-tier for BigQuery, be careful to stay under the limits and you won't pay anything). Pay attention to how your queries change, particularly with more complex queries. Notice that different kinds of queries are faster or slower between databases. Get used to reading the SQL reference documentation for each of these databases. If you need something more guided, there are plenty - literally *thousands* - of free SQL resources on the internet. There's nothing wrong with following a *free* SQL introduction course, but always challenge what you learn by applying it to a different database. You will have time to specialise in a specific database later in your career, now is not the time. **If you're in the UK, the UK Gov is sponsoring a whole bunch of entry-level bootcamps across loads of sectors. One of the biggest areas of funding is Data Engineering. I can't vouch for these bootcamps, some look better than others, but they are *free*, so if you're eligable, why not? https://www.gov.uk/guidance/find-a-skills-bootcamp/** ## What if I already know SQL? If you're coming from a Data Analytics background, or any other role where you're already reasonably comfortable with SQL, then you have it easy. Anyone with this background should be able to start looking for an entry-level or junior Data Engineering role. Now, if you're sitting on 15 years of experience as an Analyst, going back to a junior role might not be something you're willing to do - but that's a different discussion. ## What's next? What about Python? Pandas? dbt? Rust? Airflow? Spark? Later. These are all things you can learn on the job *if the job even needs them*. Go get your first data job. I'm not going to tell you it will be easy. Lots of people struggle to find the right entry-level job in all fields of engineering. But when you land it, make it your primary goal to absorb the knowledge from your new colleagues. Learn something every single day. When the learning stops, move on. Use what you've learnt to get a pay bump and find new people to learn from. Rinse and repeat. That's your roadmap. Everything else comes later. Go get your hands dirty. ### šŸŒ¶ļø A quick rant šŸŒ¶ļø *Unfortunately, I see a lot of bad advice handed out. Now, much of it is others just innocently sharing an opinion, but, often there is a clear financial incentive behind it. Vendors who want their tools to be the "baseline" to enter the industry. So-called "influencers" who take money from those vendors, or want to convince you to buy their Data Engineering bootcamps. Because, if data engineering is spooky and complicated, you're more inclined to buy their "land your first job in 90 days" course, right?* *Now, there's nothing wrong with vendors advertising their tools, or individuals creating genuinely helpful content that earns themselves a living. Tools have a place in the world, as do creators. But certain bad actors target junior and entry-level engineers who don't have the experience to identify blatant bullshit from real advice. Be wary.* --- # The (lacking) future of ETL Source: https://alasdairb.com/posts/the-lacking-future-of-etl Published: 2023-03-07 ETL provides no real value, it's just really expensive glue. So-called "Zero-ETL" has the potential to free up so much $$$ that can be put towards something actually useful. That said, we are a long, long way off from this becoming a reality. ETL is going to be around for a long while yet. I think we'll see stepped adoption: smart data vendors will begin to build "ETL" into their products, just as a value-add feature, simplifying their users' stacks and enabling easier adoption & consumption of their tool. Slower/larger vendors will either buy the ETL vendors in the market, or create a managed service from FOSS tools. They'll package it up to heavily incentivise it over anything else. The ETL vendors that don't get bought will struggle to convince people that they need to part with their cash for something that should just be a feature and doesn't provide any value of its own. Adjacent, weā€™ll see more adoption of common backends, storage layers, protocols (+ a bunch of new ones, some good, some just cashing in on the hype train). Using common storage layers, data formats, table structures (Apache Iceberg , Delta Lake), and using common interfaces (duckdb, Apache Arrow). The developers of these systems will be freed from re-building yet another data format, ser/des, network IPCs, and can focus on building the bits that actually "do something different" or "do something useful". The developers using these systems will be freed from needing to learn yet another ETL tool, building pipelines, maintaining more infra...and can focus on making their data valuable. The business will be freed from the ever-increasing spending bloat of ETL, and can instead invest that $$$ in properly utilising, or adopting new, tools that let them do new things. Again, we're a long way off from this being a reality, and there's going to be a lot of marketing noise from vendors that try to convince you that they have already have Zero-ETL (they don't). But it's an exciting vision of the future. Final thoughts: duckDB, Apache Arrow and Apache Iceberg are going to be core to the future of data tooling. The storage layer is ripe for innovation. Each cloud vendor has a blob storage service, and theyā€™re all pretty old and have limitations that arenā€™t keeping up with everything else (particularly around speed). There needs to be innovation here, ideally with a standardized API. --- # My take on Developer Experience Source: https://alasdairb.com/posts/my-take-on-developer-experience Published: 2022-11-21 Developer Experience does not mean ā€˜does this look prettyā€™ nor ā€˜does this solve a problemā€™. It means ā€˜does this help you solve the problemā€™. The latter two can sound exchangeable but they are different. There are many tools out there that can, technically, solve the problem. They have the features & capabilities to do it. But does it actually \*help\* you to solve the problem? I have had the pleasure of using many tools that \*technically\* could solve a problem, but I had to fight against the tool to make it work - they sure didnā€™t \*help\* me. Developer Experience needs to focus solely on helping the developer achieve their goal. Many tools just donā€™t care about experience, while others do care but get it wrong. I have seen a few different approaches, but in general Iā€™ve noticed two styles: ā€˜feature basedā€™ or ā€˜solution basedā€™. In my opinion, taking a feature-based approach to DevEx leads you to what I would describe as ā€˜pretty but shittyā€™. This approach agonizes over the experiences of each individual feature, ensuring that just using that one particular feature is beautiful. However, this approach tends to completely ignore that this feature is only one small part of how the developer is building a solution. It ignores the fact the developer needs to not only find the feature, but find it at the right time and in the right context to understand how/why/when to use it. It ignores that building a solution is a journey; a story with a start, middle and end. Maybe even a Shyamalan twist or two somewhere. The ā€˜solution basedā€™ approach instead focuses on guiding the developer towards their desired outcome. It takes a step back from individual features and looks at the bigger picture. How do we take the developer from their starting point A to ending point C? DevEx should be absolutely aligned with the original mission of your tool; What problem is your tool solving? The best Developer Experiences leave your users believing that your tool is \*the\* natural solution to the problem, rather than just one possible way to solve it if you think about it a bit. ā€¦perhaps this is one reason why so many get this kind of DevEx wrong; nailing the ā€˜whyā€™ behind your tool is much harder than it sounds. [Vercel](https://www.linkedin.com/feed/#) is one example of a company that absolutely nailed the Developer Experience for their product's core goal: "Deploy frontend projects". The cognitive load required to achieve that goal with Vercel is probably the lowest I have ever come across for any developer tool. At no point in the journey do you ask 'how do I solve this problem with Vercel?' you simply know 'Vercel solves this problem'. That is the goal of DevEx. --- # Mixing it up with Tinybird.co Source: https://alasdairb.com/posts/mixing-it-up-with-tinybird-co Published: 2022-06-03 I've had a brilliant 2-and-a-bit years with Cloudera, but in a surprising turn of events, I've decided it's time for something a little bit different. I left Cloudera at the at the end of May, and in June I will be starting at Tinybird in their fledgling UK team. Tinybird is a new SaaS company, with around 40 people, originally formed in Spain. They're building a platform that is the embodiment of which I have been advocating for several years now - the productisation of real-time data. Streaming & real-time are similar, but different beats, and both are pretty hard - but new players, like Tinybird, are starting to make it easier. We're at the stage where many companies have adopted Kafka/Kineses/Pulsar or some kind of messages platform; we've gotten good at building streams, but it turns out that its still pretty hard to actually _do something_ with it. Streaming analytics (or stream processing) is one such _something_, and its a very exciting space that is growing in populatiry, but still has a high bar to entry - Apache Flink is the engine _de jour_ for this, and while it has had a lot of success, many find it incredibly difficult to be productive with it (shoutout to [Decodable](https://www.decodable.co/) who are making Flink more accessible). There is also ksqlDB from Confluent, but as they say, when you only have a hammer, everything looks like a nail. Anyway, streaming analytics is the idea that we query data in-flight, allowing us to find patterns over windows of time and gain insight as the data is generated. It's very cool, it's very hard, and it has tremendous value. Another _something_ is what is often just called 'real-time data'. In real-time data, we are less concerned with running constant analytics over a stream, and instead we want to make our data available to our applications as fast as possible. Typically, these are interactive applications with strict latency, freshness and concurrency requirements. Think of the 'who viewed my profile' notifications on LinkedIn - this is real time data made available to millions of active users. Tinybird is building a platform for the latter - how do we make our data, wherever it comes from, available wherever it needs to be, as fast as possible - and do that with minimal cognitive load on developers. I think Tinybird has cracked it with a great technical product and a fantastic developer experience. I'm excited to get started with Tinybird very soon, where I'll be continuing in a technical Customer Success role, getting my hands dirty and making the next generation of real-time data products. --- # Introducing dbt-impala Source: https://alasdairb.com/posts/introducing-dbt-impala Published: 2022-03-31 As part of my new role at Cloudera, I have been looking at a tool called [dbt](https://www.getdbt.com/). To many, dbt needs no introduction - it has not just made a splash in the data ecosystem, it has come in riding a tsunami of positivity, and for good reason! Just looking at it's [star history](https://www.star-history.com/#dbt-labs/dbt-core&Date) we can see it's adoption is only getting faster and faster - I don't see that changing any time soon. It is quickly becoming synonymous with the Modern Data Stack. ![](./images/introducing-dbt-impala-1.png) ## What is dbt? If you haven't come across dbt before, it doesn't always sound very exciting - I've heard some folks compare it to saving SQL scripts in GitHub, which I suppose isn't a million miles off, but there's a lot more to it than that! To quote dbtLabs: > dbt also enables analysts to work more like software engineers > > https://docs.getdbt.com/docs/introduction dbt essentially allows you to create your analytics projects using the same processes that software projects have been using for years now. In software, there's a lot of concepts that every single team knows and uses; - collobrative development - version control - don't repeat yourself (DRY) - documentation - testing How many analytics projects utilise these things? In my experience, it's very few - and that's not because they don't see the value of it, it's because there has been very little capability to reliably implement these ideas in the analytics space. Many analytics teams have lived this pain; work ends up silod in a member's 'user' space, queries end up being written more than once by different members (often with varying levels of optimisation), and after a year of hard work - there is little, if any, documentation to bring on new team members. dbt finally enables analytics teams to fully leverage these concepts in their workflows. [Read more straight from the horses mouth](https://docs.getdbt.com/community/resources/viewpoint). ## dbt and Impala Cloudera provides several different engines that analytics teams can use - Impala, Hive and Spark. These engines serve different use cases & target users, though there is some overlap. Impala is a fan-favourite for ad-hoc, low-latency analytics over vast datasets that go up to & beyond the petabyte scale. We saw a lot of demand for using dbt with Impala to assist analytics teams in re-structuring how they organise their projects. As such, our small and brand new team ([we're hiring btw](https://www.cloudera.com/careers.html)!) has been hard at work building out this functionality, to bring dbt to Apache Impala & Cloudera Data Platform. In just a few weeks, we have gone from a blank canvas to a functional adapter that is already being used in the real world. We are targetting a new release every week to continue adding the standard dbt features that users expect. The good news - it's all 100% free & open source! ## Get dbt-impala You can find the dbt-impala on [GitHub here](https://github.com/cloudera/dbt-impala). It's available via [PyPi](https://pypi.org/project/dbt-impala/1.0.1/) & can be installed with pip `pip install dbt-impala` --- # The Cloudera Innovation Accelerator & me Source: https://alasdairb.com/posts/the-cloudera-innovation-accelerator-me Published: 2022-03-31 I recently took on a new role at Cloudera as a Senior Developer Advocate in the newly formed Cloudera Innocation Accelerator. This exciting new team sits between Cloudera and the wider data community, looking at new & ermeging technologies that are in-demand & have real potential to provide value to Cloudera's customers. In this role, I'll be keeping a finger on the pulse of the whole data ecosystem, trying to find any new projects that are getting people excited, assessing how they might apply to Cloudera's products & customers, and getting to share the final results back with the community & our customers - hopefully bringing a little of that excitement along with me! It's a fantastic role, and I could not be happier with the team I am working with - a diverse array of individuals that bring their own unique mix of brilliance and personality. I can't wait to share what we have been working on - stay tuned! --- # Getting started with NiFi's ScriptedReader Source: https://alasdairb.com/posts/getting-started-with-nifis-scriptedreader-by-processing-key-value-pairs-part-1 Published: 2021-09-21 Records have become an integral part of working with NiFi since their introduction on May 8th, 2017 with the release of [NiFi 1.2.0](https://cwiki.apache.org/confluence/display/NIFI/Release+Notes#ReleaseNotes-Version1.2.0). They allow us to treat our data as more than just a bunch of bytes, giving NiFi the ability to better understand and manipulate common data formats used by other tools. However, there's always some tool out there that decides to do things a bit _differently_. JSON? AVRO? CSV? Nah. Here's a data format we came up with after one _really_ weird weekend in Vegas. So, while NiFi has out of the box support for at least 8 of the most common formats, how do we handle the _not so common_ ones? Well, part of the 1.2.0 release note give us a hint: --- > "For those that like to write new capabilities on the fly using scripting languages you can build your own reporting tasks, record readers, and writers using various scripting langauges..." --- This is alluding to the **[ScriptedReader](https://nifi.apache.org/docs/nifi-docs/components/org.apache.nifi/nifi-scripting-nar/1.14.0/org.apache.nifi.record.script.ScriptedReader/index.html)**, a pretty powerful feature that allows us to hook in to NiFi's Record reading capabilities and define our own logic for making sense of our data. This post is going to walk through creating our first ScriptedReader, just scratching the surface of the capabilities of this feature. I have seen several NiFi users ask the question "How should I process Key Value Pairs?" so taking the two-birds-one-stone approach, I'm going use that as the example here. Without further ado, let's jump in to it. ## Test Flow First, you need a working install of NiFi - if you don't already have one, I wrote [a post about getting one set up](/posts/installing-nifi-1-14-0-on-linux-non-production). When you're in NiFi, let's set up a basic Flow that generates some sample data and tries to parse it as a series of Key Value Pairs (KVPs). First, add a **GenerateFlowFile** processor. We'll use this to generate our sample data. Configure the processor as follows: - On the **Scheduling** tab, set the **Run Schedule** to **10 sec** - On the **Properties** tab, set the **Custom Text** to: ``` name:dennis age:45 job:veterinarian name:dee age:45 job:actress name:mac age:44 job:sheriff name:charlie age:45 job:lawyer ``` ![](./images/getting-started-with-nifis-scriptedreader-by-processing-key-value-pairs-part-1-1.png) GenerateFlowFile Config After this, add an **UpdateRecord** processor, then configure the **Properties** tab as follows: - For **Record Reader**: - Select **Create new service...** from the drop down - Select **ScriptedReader** as the **Compatible** **Controller** **Service** - Enter **KVP Scripted Reader** as the **Controller Service Name** - For **Record Writer**: - Select **Create new service...** from the drop down - Select **JSONRecordSetWriter** as the **Compatible** **Controller** **Service** - Enter **JSON Record Writer** as the **Controller Service Name** - Set **Replacement Value Strategy** to **Literal Value** - Add a dynamic property with the name **/location** and a value **philly** ![](./images/getting-started-with-nifis-scriptedreader-by-processing-key-value-pairs-part-1-2.png) UpdateRecord Config Next, drag on two **Funnels**. Now connect the **GenerateFlowFile** output to the **UpdateRecord**, then connect the **UpdateRecord's** failure relationship to one **Funnel** and the success relationship to the other **Funnel.** The Flow should look like this screenshot: ![](./images/getting-started-with-nifis-scriptedreader-by-processing-key-value-pairs-part-1-3.png) Complete Test Flow That's all we need to demonstrate our custom Reader. You can find a link to download this Flow from GitHub at the bottom of the post. ## Development Environment We are going to be using the Groovy language to create our ScriptedReader. A variety of languages are supported, but Groovy is the most robust. Developing a script inside NiFi is cumbersome and error prone - it's significantly easier to build our script outside of NiFi where we can take advantage of a proper editor with syntax highlighting, linting etc. and then copy our script back into NiFi. My editor of choice is [Visual Studio Code](https://code.visualstudio.com/), and there is a plugin for Groovy called [Groovy Lint, Format and Fix](https://marketplace.visualstudio.com/items?itemName=NicolasVuillamy.vscode-groovy-lint). It is also possible to run Groovy code locally to test functionality, and for this we need a Groovy SDK. The easiest way to get set up is using [SDKMAN](https://sdkman.io/) to handle downloading, installing and activating SDKs for us. Follow the [install instructions for SDKMAN](https://sdkman.io/install/). With SDKMAN installed, we can use it to install Groovy. NiFi 1.14.0 uses Groovy 2.5.4, so let's install that version. ```bash sdk install groovy 2.5.4 ``` With that installed, you can now use the `groovy` command to execute Groovy scripts. Verify that you can run `groovy`: ```bash groovy -v ``` To execute a Groovy script you would use something like the following: ```bash groovy /path/to/script.groovy ``` ## Building the script locally [Find the full code sample here.](https://github.com/sdairs/nifi-snippets/blob/main/flows/scripted-reader-key-value-pair/local-kvp-testing.groovy) Our local system has no knowledge of the NiFi internals, but we can build the main logic of our Reader outside of NiFi to allow us to more quickly test & iterate. Fundamentally, our script needs to be able to: 1. Read a line of text 2. Split the line up in to Key Value Pairs 3. Split each Key Value Pair into a Key and a Value 4. Return an object containing the Keys and Values 5. Move on to the next line, repeat Let's start by creating the logic for these basic steps first. Please keep in mind that this is only a basic implementation of parsing KVPs for demonstration purposes, so there's a lot it won't do. To start the script, we need our test data as shown at the start of this post. We'll use a multiline String to contain the test data. ```java test_data = '''name:dennis age:45 job:veterinarian name:dee age:45 job:actress name:mac age:44 job:sheriff name:charlie age:45 job:lawyer''' ``` Next, we need a way to read our string line by line, and do _something_ with each line. So let's add a little loop that splits by a newline, calls a function on each line, and pushes the results to an array. We'll also print the array so we can see the result. Notice that we are calling a method called **parse()** that we have not implemented yet. ```java records = [] test_data.split('\\n').each { line -> records.add(parse(line)) } print(records) ``` Let's create the **parse()** method. This method will contain the functionality we want to bring into NiFi. So, it needs to accept an input line and return us the set of Key Value Pairs. We'll also do a basic check for the line being null. Our return type is _Map_ because we have a Key, which will always be a String, and a Value which could be any kind of data. This is also the format NiFi will expect for Records when we port this to NiFi. ```java Map parse(String line) { if (line == null) { return null } ...rest of the code here... } ``` Now we need to add the basic parsing logic to our **parse()** method. Each call of the method is getting an individual line of the sample data, e.g. ``` name:dennis age:45 job:veterinarian ``` So, we need to split this line up in to the individual KVPs, which in our basic example is just separated by a single space (meaning this example won't handle Values with spaces!) ```java List kvps = line.tokenize(' ') ``` The Groovy **tokenize()** method splits the line by a delimiter (we set it to a single space) and returns an iterable List of the splits, e.g. for the first line in the data, we have 3 splits (our KVPs): ``` [name:dennis, age:45, job:veterinarian] ``` Now that we have the set of KVPs, we need to split each Key from the Value (separated by the colon) and collect the results into our Map, to represent our Record. ```java Map recordValues = [:] kvps.each { row -> String key = '' String value = '' (key,value) = row.tokenize(':') recordValues.put(key, value) } return recordValues ``` This is the basic logic of our KVP Reader. We can execute the code locally to test with: ```bash groovy ./nifi-snippets/flows/scripted-reader-key-value-pair/local-kvp-testing.groovy ``` Which gives us the following output: ``` [ [name:dennis, age:45, job:veterinarian], [name:dee, age:45, job:actress], [name:mac, age:44, job:sheriff], [name:charlie, age:45, job:lawyer] ] ``` Not the most interesting on its own, but in NiFi, each Map in this List is going to be a Record. So let's start getting our code ready for NiFi. ## Building the script for NiFi [Find the full code sample here.](https://github.com/sdairs/nifi-snippets/blob/main/flows/scripted-reader-key-value-pair/kvp-scripted-reader.groovy) When creating a ScriptedReader, NiFi is expecting it to have two things: a **[RecordReaderFactory](https://javadoc.io/static/org.apache.nifi/nifi-record-serialization-service-api/1.14.0/org/apache/nifi/serialization/RecordReaderFactory.html)** and the associated **[RecordReader](https://www.javadoc.io/doc/org.apache.nifi/nifi-record/latest/org/apache/nifi/serialization/RecordReader.html)**. Let's start with the **RecordReaderFactory** which, in our case, is just going to implement the required method **createRecordReader()** and return an instance of our RecordReader. You'll see that the **createRecordReader()** takes a bunch of parameters, but in this example we're only concerned with the `**InputStream inputStream**` which contains the actual stream of data that forms the content of the FlowFile, which we pass to our RecordReader. ```java class KVPReaderFactory extends AbstractControllerService implements RecordReaderFactory { public KVPReaderFactory() { } public RecordReader createRecordReader(final Map variables, final InputStream inputStream, final long inputLength, final ComponentLog componentLog) throws IOException { return new KVPReader(inputStream) } } ``` Next, we need to add our **RecordReader** which we'll call **KVPReader**. For the RecordReader, we must implement the **constructor**, and 3 methods: **getSchema()**, **close()** and **nextRecord()**. Our **constructor** is simply going to pass the **inputStream** from the RecordReaderFactory to a BufferedReader, which will handle consuming the byte stream of FlowFile content, giving us consumable lines of text. In this example we aren't going to cover Schemas in detail, but the **getSchema()** method must be implemented and return a schema. When parsing KVPs, we will generate a basic Schema where every value is represented by a String. Next, we must also implement the **close()** method which properly closes out the BufferedReader that is handling our input stream. Finally, the **nextRecord()** method is called every time the Reader has finished reading & returned one Record, and must process the next one. It is this method that will implement the logic that we created previously in our local script. ```java class KVPReader implements RecordReader { private final BufferedReader bufferedReader public KVPReader(InputStream input) { bufferedReader = new BufferedReader(new InputStreamReader(input)) } public Record nextRecord(final boolean coerceTypes, final boolean dropUnknownFields) throws IOException, MalformedRecordException { .... } @Override public void close() throws IOException { bufferedReader.close() } @Override public RecordSchema getSchema() { return schema } } ``` Looking specifically at **nextRecord()** we need to bring in the logic we created earlier, and add a few bits to make it work in NiFi. The first thing to notice is that, instead of just returning **Map** the method is actually returning a [**Record**](https://www.javadoc.io/doc/org.apache.nifi/nifi-record/latest/org/apache/nifi/serialization/record/Record.html). This **Record** object is made up of our **Map**, which contains the data, in addition to a **RecordSchema** that defines field names and data types. Next, we aren't just setting up a String with our test data in, we need to actually consume the FlowFile content from our BufferedReader. ```java final String line = bufferedReader.readLine() ``` This gives us a single line of KVPs, just as we had before in our **parse()** method. The rest of the **nextRecord()** method is very similar to our local script, with a few changes. We need a new variable to contain a List of [RecordField](https://www.javadoc.io/static/org.apache.nifi/nifi-record/1.14.0/org/apache/nifi/serialization/record/RecordField.html) objects - this list will be used to create the [RecordSchema](https://www.javadoc.io/static/org.apache.nifi/nifi-record/1.14.0/org/apache/nifi/serialization/record/RecordSchema.html). ```java List recordFields = [] ``` Inside our `**each**` loop, we need to now populate the **`recordField`** list, so we create a **RecordField** for each KVP and add it to the list. A **RecordField** needs a field name and a [**RecordFieldType**](https://www.javadoc.io/static/org.apache.nifi/nifi-record/1.14.0/org/apache/nifi/serialization/record/RecordFieldType.html) to represent the data type. For the name, we will reuse the Key of the KVP. For the data type, we are simply going to use Strings for this example. ```java recordFields.add(new RecordField(key, RecordFieldType.STRING.getDataType())) ``` Next, we need to create an instance of a **RecordSchema**, so we'll keep it basic with the [**SimpleRecordSchema**](https://www.javadoc.io/static/org.apache.nifi/nifi-record/1.14.0/org/apache/nifi/serialization/SimpleRecordSchema.html) and pass it our list of field types. ```java SimpleRecordSchema schema = new SimpleRecordSchema(recordFields) ``` Finally, we need to create & return our **Record**. You can write your own class to implement the Record interface, but we'll just use the ready made [MapRecord](https://www.javadoc.io/static/org.apache.nifi/nifi-record/1.14.0/org/apache/nifi/serialization/record/MapRecord.html). A MapRecord takes in a **RecordSchema** and a **Map** of the values. ```java return new MapRecord(schema, recordValues) ``` The rest of the code in the method is identical to our original local version. See the GitHub repo for the complete code. ## Run it in NiFi With our code complete, we now need to test it in NiFi. Return to the NiFi flow and enter the configuration for the **ScriptedReader** called **KVP Record Reader**. Copy the complete script and paste it into the **Scipt Body** property. Ensure that **Script Engine** is set to **Groovy**. ![](./images/getting-started-with-nifis-scriptedreader-by-processing-key-value-pairs-part-1-4.png) ScriptedReader Config Save this config, and enable both the **ScriptedReader** and **JsonRecordSetWriter** by clicking the **Enable** button. ![](./images/getting-started-with-nifis-scriptedreader-by-processing-key-value-pairs-part-1-5.png) Enable the Record Reader & Writer You should now be able to start both the **GenerateFlowFile** and **UpdateRecord** Processors. A FlowFile should end up in the **Success** queue, containing a JSON representation of our test data plus the additional **location** field. ![](./images/getting-started-with-nifis-scriptedreader-by-processing-key-value-pairs-part-1-6.png) Open the **Success** queue and view the content of the FlowFile. ![](./images/getting-started-with-nifis-scriptedreader-by-processing-key-value-pairs-part-1-7.png) JSON Data That's it! Find the [complete scripts & the flow definition on the GitHub repo for this post](https://github.com/sdairs/nifi-snippets/tree/main/flows/scripted-reader-key-value-pair). --- # NiFi Terminology Basics Source: https://alasdairb.com/posts/nifi-terminology-basics Published: 2021-09-05 This post is going to cover the basic terminology that you'll need to know for working with NiFi. ### Canvas The Canvas is the free grid space on which you create your Flow. It is accessed through the NiFi Web UI. You build Flows by dragging Processors on to the Canvas. ### Controller Service A Controller Service is a encapsulation of functionality that is consumed or used by a Flow or Processor and does not operate independently. The same Controller Service can be shared by multiple Processors/Flows. An example is the **AvroRecordSetWriter**, which encapsulates the functionality for writing Records with a given Avro schema, but does not do anything on it's own - it must be consumed by a supporting Record Processor. ### Flow A Flow is the combination of Processors, Controller Services, Funnels, Ports, etc. that are connected through their relationships to move and/or process some data. A Flow is built and viewed through the NiFi Web UI on the Canvas. ### FlowFile A FlowFile is the unit of 'data' in NiFi. It includes both the _content_ of the data and it's associated _attributes_ (metadata)**.** FlowFiles are created, and consumed, by Processors, which can read and/or modify either the content or the attributes. ### FlowFile Attribute A FlowFile Attribute is a single piece of metadata attached to a FlowFile. These Attributes are stored in memory and are intended for smaller pieces of information that describes the data, rather than the data itself. ### FlowFile Content FlowFile Content is used for the underlying data that the Flow is operating on. It can be any kind of data, such as textual JSON data or binary data like images and videos. FlowFile content can contain very large pieces of data that is not appropriate to keep in FlowFile Attributes. FlowFile Content is kept on disk. ### Input Port An Input Port is used to provide an input relationship to a Process Group, allowing for Processors _outside_ of the Process Group to send FlowFiles to Processors _inside_ the Process Group. Input Ports can be _local_ or _remote_. A _local_ Input Port can only be referenced by Process Groups that run in the same NiFi cluster as the Input Port and can attach directly in the Flow. A _remote_ Input Port allows for Remote Process Groups to connect to the Input Port over a network, without needing to be directly connected inside the Flow or running in the same NiFi cluster. ### NiFi Registry NiFi Registry is a companion project of NiFi that provides Git-like version control for NiFi flows. It runs as a seperate service to NiFi, providing it's own Web UI for management. NiFi connects to a NiFi Registry service through a Registry Client configured through the NiFi Web UI. ### Output Port An Output Port is used to provide an output relationship from a Process Group, allowing Processors _inside_ the Process Group to send FlowFiles to Processors _outside_ of the Process Group. ### Parameter A Parameter is a Process Group level variable that is statically configured inside a Parameter Context. Parameters can optionally contain sensitive values. Parameters are referenced by name inside a Processor's configuration, using the _#{name}_ syntax. A Paramater's value is evaluated at the time the referencing Processor is started, and the Processor(s) must be stopped before the Parameter's value can be changed. ### Parameter Context A Paramater Context is a container for one or more Parameters. A Parameter Context can be attached to one or more Process Groups, but a Process Group may only have one Parameter Context. A Processor can only reference Parameters in the Parameter Context that is attached to it's Process Group. This is the preferred method for static variables in NiFi, with the Variable Registry being considered depricated. ### Process Group A Process Group is a nested container for Flows. Process Groups can be though of as similar to Directories or Folders in a File System. Each Process Group gets it's own Canvas. The top level Process Group is known as the Root Process Group, which is the first Canvas presented to the user when accessing the NiFi Web UI. Process Groups can be used to organise Flows in to smaller areas of focus. ### Processor A Processor is the core functional component in NiFi. Each Processor provides a specific piece of functionality, typically operating on either the Content or Attributes of a FlowFile. Processors can optionally accept input FlowFiles. Processors always produce an output that can be consumed by other Processors. A Processor can have one or more output Relationships that can be connected to other processors. ### Remote Process Group A Remote Process Group is similar in concept to a Process Group; It is a logical container around a Flow. However, a Remote Process Group does not have to be running on the local NiFi cluster. It can be used to pass FlowFiles to a Flow that is running on an different, external NiFi cluster over a network. The Remote Process Group must have a Remote Input Port that allows external input. ### Root Canvas The Root Canvas is the first Canvas that is presented to the user when accessing the NiFi Web UI. It is the Canvas of the Root Process Group. ### Variable Registry The Variable Registry is a container for one of more static variables that are set outside of a Flow. They cannot be modified by Processors, and their value is evaluated at the time of a Processor being started. The Variable Registry is considered deprecated and Parameters should be used instead. --- # Installing NiFi 1.14.0 on Linux (non-production) Source: https://alasdairb.com/posts/installing-nifi-1-14-0-on-linux-non-production Published: 2021-09-04 This post is going to walk through installing the latest release of NiFi on a Linux host (Fedora). This latest release is version 1.14.0 and an important change is that NiFi is now **[secure by default](https://bryanbende.com/development/2021/07/19/apache-nifi-1-14-0-secure-by-default)**. This means that HTTPS is enabled out of the box (with self signed certs), user authentication is required and the default binding is to localhost/127.0.0.1. ## Installing on Linux The Linux host I am using is running Fedora Workstation 34, but the steps are pretty much the same on most flavours of Linux. If you're on Windows, do yourself a favour and switch to Linux. ### Installing Java First of all, you need to have an install of Java. NiFi 1.14.0 supports Java 8 and 11, so I will use OpenJDK11. On Fedora 34, you can find the available versions of OpenJDK by running ```bash dnf search openjdk ``` Use _apt_ on Ubuntu or _yum_ on centos/rhel 7. I'll install OpenJDK11 using the following ```bash dnf install java-11-openjdk-devel ``` Using the _\-devel_ package is not strictly necessary. Let's check that it has installed properly. Run the following ```bash java -version ``` And you should see something like ```bash openjdk version "11.0.12" 2021-07-20 OpenJDK Runtime Environment 18.9 (build 11.0.12+7) OpenJDK 64-Bit Server VM 18.9 (build 11.0.12+7, mixed mode, sharing) ``` That's Java good to go. ### Installing NiFi Let's start by making a directory on our machine for NiFi. ```bash mkdir ~/nifi cd ~/nifi ``` Now we need to download the NiFi binaries. The downloads page is [here](https://nifi.apache.org/download/). The mirrors page for the 1.14.0 release is [here](https://www.apache.org/dyn/closer.lua?path=/nifi/1.14.0/nifi-1.14.0-bin.tar.gz). Start by downloading the [nifi-1.14.0-bin.tar.gz](https://apache.mirrors.nublue.co.uk/nifi/1.14.0/nifi-1.14.0-bin.tar.gz) from a mirror of your choice. You can either do this in your browser, or through the command line. Have the _tar.gz_ file in your nifi folder. The file size is 1.3GB. ```bash wget https://apache.mirrors.nublue.co.uk/nifi/1.14.0/nifi-1.14.0-bin.tar.gz ``` Download the [SHA256 hash](https://downloads.apache.org/nifi/1.14.0/nifi-1.14.0-bin.tar.gz.sha256) to the same directory. ```bash wget https://downloads.apache.org/nifi/1.14.0/nifi-1.14.0-bin.tar.gz.sha256 ``` Now compare the hash to the file we downloaded to ensure we've got the right thing. ```bash echo "$(cat nifi-1.14.0-bin.tar.gz.sha256) nifi-1.14.0-bin.tar.gz" | sha256sum --check ``` You should see this if all is good. ```bash nifi-1.14.0-bin.tar.gz: OK ``` If it doesn't return OK, your download might be corrupt or incomplete. Next, we need to unpack the archive file. ```bash tar -xzf nifi-1.14.0-bin.tar.gz ``` When that completes, you'll have a new directory called **nifi-1.14.0.** Enter the new directory. ```bash cd nifi-1.14.0/ ``` The contents of the directory should look like this ```bash LICENSE NOTICE README bin conf docs extensions lib ``` The **conf** directory contains configuration files, while **bin** contains scripts for starting or interacting with the NiFi process. At this point, we can start NiFi using the following: ```bash ./bin/nifi.sh start ``` This will produce output similar to: ```bash nifi.sh: JAVA_HOME not set; results may vary Java home: NiFi home: /user/nifi/nifi-1.14.0 Bootstrap Config File: /user/nifi/nifi-1.14.0/conf/bootstrap.conf ``` This won't tell us if it has started or if there were any issues, so you will want to tail the nifi-app.log to watch for successful starts, error messages, and the auto-generated credentials. ```bash tail -f -n 100 ./logs/nifi-app.log ``` The line that you want to be watching for is this: ```bash 2021-09-04 18:09:55,481 INFO [main] org.apache.nifi.web.server.JettyServer https://127.0.0.1:8443/nifi ``` Which tells us that NiFi successfully started and the UI is now available. This line contains the URL that the NiFi UI is available on. Pay careful attention to the address, as by default it is only available on 127.0.0.1, which means you'll only be able to reach the UI from **the same host that NiFi is running on**. No other host on your network can reach this address. You will also want to watch out for the following two lines, which contain the auto-generated admin credentials for logging in to NiFi. ```bash Generated Username [96f17a45-5fa6-4ebe-bb3d-13dbdf836867] Generated Password [dUYc1Pu835IgXO3txn1MqM2/YLNC1AYA] ``` To verify that NiFi is available, either browse to **https://127.0.0.1:8443/nifi** in your browser (again, this is ONLY on the same machine that is running NiFi) OR use curl to check from the command line. ```bash curl -sSLk -D - https://127.0.0.1:8443/nifi -o /dev/null ``` You will see a response similar to this, which means all is good (we get a 302 redirect into a 200 OK response, this is normal). ```bash HTTP/1.1 302 Found Date: Sat, 04 Sep 2021 17:19:27 GMT Location: https://127.0.0.1:8443/nifi/ Content-Length: 0 Server: Jetty(9.4.42.v20210604) HTTP/1.1 200 OK Date: Sat, 04 Sep 2021 17:19:27 GMT X-Frame-Options: SAMEORIGIN Content-Security-Policy: frame-ancestors 'self' X-XSS-Protection: 1; mode=block X-Content-Type-Options: nosniff Strict-Transport-Security: max-age=31540000 Content-Type: text/html;charset=utf-8 Vary: Accept-Encoding, User-Agent Transfer-Encoding: chunked Server: Jetty(9.4.42.v20210604) ``` At this point, NiFi is perfectly usable - if you can reach the web UI, you will see a logon screen and you can use the auto-generated credentials that we saw in the nifi-app.log to login. If that's all you need, you're done. However, if you need NiFi accessible from other machines, perhaps because the machine running NiFi is a headless server (like mine) then we need to stop NiFi, change the bind host and start it up again. First, stop NiFi using the same nifi.sh script as before: ```bash ./bin/nifi.sh stop ``` Once stopped, we need to edit the **nifi.properties** file which is in the **conf** dir. ```bash vim ./conf/nifi.properties ``` Once in the file, you're looking for the following lines: ```bash nifi.web.https.host=127.0.0.1 nifi.web.https.port=8443 ``` As you can see, we are binding to 127.0.0.1. Change this address to the one you want NiFi to be available on. I will use 0.0.0.0 which makes NiFi available on all interfaces. ```bash nifi.web.https.host=0.0.0.0 nifi.web.https.port=8443 ``` Start NiFi back up again. ```bash ./bin/nifi.sh start ``` Now from another host on the network you can reach NiFi on the public address/name of the NiFi host. In my case, the NiFi host is on 192.168.2.2 (**https://192.168.2.2:8443/nifi/**). That's it. NiFi is installed and accessible on the network. ## Common Issues If you see this response in curl, it's because you forgot the **\-k** flag, which ignores the self-signed certificate warning. ```bash curl: (60) SSL certificate problem: self signed certificate More details here: https://curl.se/docs/sslcerts.html curl failed to verify the legitimacy of the server and therefore could not establish a secure connection to it. To learn more about this situation and how to fix it, please visit the web page mentioned above. ``` If you see this garbled response in your browser, it's because you aren't accessing the URL on HTTPS (you **must** use **https://**) ![](./images/installing-nifi-1-14-0-on-linux-non-production-1.png) --- # S3 Event Notifications in NiFi Source: https://alasdairb.com/posts/s3-event-notifications-in-nifi Published: 2021-09-04 The desire to pull newly uploaded files from storage is common, and typically the ListFile -> FetchFile pattern has been used in NiFi to monitor for, and then pull, new files as they arrive. Taking NiFi in to the cloud means that we're often working with cloud object storage, like Amazon S3. While we can use a similar pattern with ListS3 -> FetchS3Object, we could instead build a more robust flow by reacting to [Amazon S3 Event Notifications](https://docs.aws.amazon.com/AmazonS3/latest/userguide/NotificationHowTo.html). This feature enables S3 to publish messages to an Amazon SNS topic to notify downstream services of changes in the S3 bucket. This way, rather than having to actively maintain a state of the directory in NiFi, we can just wait to be told that something has changed, and choose how we repond to it. The Event Notification feature writes to SNS and, while NiFi can poll SNS topics with GetSNS, it's recommended to have SNS persist messages in to an SQS queue. By doing this, we can have many downstream applications (or just many flows) that can all consume the same messages independently - and we get more options to handle failures. So, the basic architecture of this goes something like: ![S3 Bucket -> SNS Topic -> SQS Queue -> NiFi GetSQS -> NiFi FetchS3](./images/s3-event-notifications-in-nifi-1.png) ## Sample Data Files For this example I will be using 2 sample files. File 1 - sample-data1.json ```json [ { "color": "red", "value": "#f00" }, { "color": "green", "value": "#0f0" }, { "color": "blue", "value": "#00f" } ] ``` File 2 - sample-data2.json ```json [ { "color": "magenta", "value": "#f0f" }, { "color": "yellow", "value": "#ff0" }, { "color": "black", "value": "#000" } ] ``` ## Setting up AWS Start by [creating an S3 bucket](https://docs.aws.amazon.com/AmazonS3/latest/userguide/create-bucket-overview.html). Then create a new standard SQS Queue. Next, create a new standard SNS Topic. Once created, add your S3 bucket to the access policy for the SNS Topic. For this, the below policy is enough. Be sure to replace ****, **** and **** with your details. ```json { "Version": "2012-10-17", "Id": "example-ID", "Statement": [ { "Sid": "example-statement-ID", "Effect": "Allow", "Principal": { "Service": "s3.amazonaws.com" }, "Action": ["SNS:Publish"], "Resource": "", "Condition": { "ArnLike": { "aws:SourceArn": "arn:aws:s3:*:*:" }, "StringEquals": { "aws:SourceAccount": "" } } } ] } ``` Now add a new subscription to the SNS Topic, which will look like this. ![](./images/s3-event-notifications-in-nifi-2.png) Once created, you need to go back to the SQS Queue and find the subscription in the SNS Subscriptions tab and confirm the subscription - tick the SNS item, click the big orange _Subscribe to Amazon SNS topic_ button, select the SNS resource and confirm. Now go back to the S3 bucket you created, and click on the **Properties** tab. Scroll down to **Event Notifications**. Create a new event notification, set a name and select the operations you want to be notified about - for this example, I will select _All object create events_. Lastly, take a note of these details: - SQS Queue URL (go to the SQS Queue in AWS and you will see the URL in the details) - IAM Role Access Key ID & Secret Access Key (If you don't already have one, go to AWS IAM and generate an Access Key for your IAM user) - The AWS Region you created your resources in (e.g. EU Ireland) Further information here on the [AWS docs](https://docs.aws.amazon.com/AmazonS3/latest/userguide/ways-to-add-notification-config-to-bucket.html). ## Setting up NiFi If you don't already have NiFi installed, I have a guide on [installing NiFi 1.14.0](/posts/installing-nifi-1-14-0-on-linux-non-production). Add a new Process Group to contain our flow, I'll call this one 's3-event-notification-in-nifi'. Right click the new Process Group and click Configure. Under **Process group parameter context** select **Create new parameter context**. Give the parameter context a name. Add 3 parameters with the details from AWS: - SQS Queue URL - Access Key ID (tick Yes for Senstive Property) - Secret Access Key (tick Yes for Senstive Property) Drag in a **GetSQS** processor to the root flow and configure the following properties (we use the parameters we just created) - Queue URL = #{SQS Queue URL} - Access Key ID = #{Access Key ID} - Secret Access Key = #{Secret Access Key} - Region = pick the region you used in AWS Drag in an **EvaluateJsonPath**. Set the following properties: - Destination = flowfile-content. Add a new dynamic property: - message = $.Message The Event Notification message contains a key called Message which contains an escaped JSON object string. Here replace the content of the FlowFile with this JSON object (now un-escaped). ![](./images/s3-event-notifications-in-nifi-3.png) Drag in another **EvaluateJsonPath**. Set the following properties: - Destination = flowfile-attribute Add a the following dynamic properties: - bucket \= $.Records\[0\].s3.bucket.name - filename \= $.Records\[0\].s3.object.key ![](./images/s3-event-notifications-in-nifi-4.png) Drag in a **FetchS3Object**. Set the following properties: - Access Key ID = #{Access Key ID} - Secret Access Key = #{Secret Access Key} - Bucket = ${bucket} - Object Key = ${filename} Drag in a Funnel. Connect the processors up like this: GetSQS (success) > EvaluateJSONPath (1, matched) > EvaluateJSONPath (2, matched) > FetchS3Object (success) > Funnel ![](./images/s3-event-notifications-in-nifi-5.png) Start all of the processors. Now go back to the S3 bucket in AWS and upload the two sample json files. Return to NiFi and wait. The SQS notification will usually take a few seconds to arrive. The Funnel at the end of the flow will end up with 2 FlowFiles containing the contents of the 2 sample files. We're done! This is an alternative strategy to using the List > Fetch pattern in NiFi, often called an 'Event Driven Fetch' pattern. [You can find this Flow on Github.](https://github.com/sdairs/nifi-snippets/tree/main/flows/s3-event-notification-in-nifi) --- # Importing individual flow definitions in NiFi Source: https://alasdairb.com/posts/importing-individual-flow-definitions-in-nifi Published: 2021-08-22 While I recommend you [Version Control & Deploy flows with NiFi Registry](/posts/nifi-in-production-nifi-registry) you can also quickly import flow definition json files straight from the NiFi GUI. This post will quickly demonstrate how. To get some json files to work with, clone my nifi-snippets repo that contains all of the NiFi flow demos that I have written about. ```bash git clone https://github.com/sdairs/nifi-snippets.git ``` Now, go to your NiFI instance & drag on a new Process Group from the top bar. ![](./images/importing-individual-flow-definitions-in-nifi-1.png) In the Process Group dialogue, hit the Browse button (it looks like two boxes with an up-arrow) inside the PG Name input field. ![](./images/importing-individual-flow-definitions-in-nifi-2.png) This will open a file browser to let you select a flow definition to import. Navigate to the clones git repo and select the json file to upload - I will upload the simple-rest-api flow by selecting simple-rest-api.json and hitting Open. ![](./images/importing-individual-flow-definitions-in-nifi-3.png) The dialogue box will show you which file is being uploaded and pre-populating the name field with the name of the file - you are welcome to change the name. Click Add. ![](./images/importing-individual-flow-definitions-in-nifi-4.png) The new Process Group will be added to the flow. ![](./images/importing-individual-flow-definitions-in-nifi-5.png) Enter the Process Group by double clicking on it, and you'll see the flow that was imported. **Note**: importing a Flow Definition does **not** enable the controller services, so in this case the HandleHTTPRequest and HandleHTTPResponse processors have invalid config warnings. ![](./images/importing-individual-flow-definitions-in-nifi-6.png) That's all there is to it. --- # Building a simple REST API in NiFi Source: https://alasdairb.com/posts/building-a-simple-rest-api-in-nifi Published: 2021-08-09 In a [previous post](/posts/enriching-records-with-lookuprecord-rest-apis-in-nifi) I discussed using REST APIs to enrich records at the time of ingest. This post will cover building the corresponding REST API that I used in that post. This will be a very simple REST API that exposes a single endpoint _**GET /api**_. This endpoint returns a 200 OK response with a small piece of JSON that changes based on the value of a URL parameter **_param1_**. Any other path or method will return a 404. First, we need a web server that accept HTTP requests. For this, we can use the **HandleHTTPRequest** processor (behind the scenes this includes a Jetty web server). Add the **HandleHTTPRequest** and double click it to enter the config window. You can customise the **Port** it listens on, but I will leave it at the default 80. You must add & enable a **HTTP Context Map** - simply select _Create new service..._ from the drop down and create a new one with default settings (then enable it). Next is **Allowed Paths** that controls which paths the API will respond to - this is a regex pattern so can include many paths, but in my case I am only allowing one (/api). Lastly, set all methods except _GET_ to _false_ (e.g. **Allow Post, Allow PUT,** etc). ![](./images/building-a-simple-rest-api-in-nifi-4.png) That's all we need to accept HTTP Requests. Hit Apply. Now, we need to return a response. Add a new **HandleHTTPResponse**. Set the **HTTP Status Code** to 200 and for **HTTP Context Map** select the same service you created for the HandleHTTPRequest. That's it, hit Apply. ![](./images/building-a-simple-rest-api-in-nifi-1.png) Now connect the **Success** relationship from **HandleHTTPRequest** to **HandleHTTPResponse**. Auto-terminate any other relationships (or add them to funnels if you want). ![](./images/building-a-simple-rest-api-in-nifi-3.png) Start the processors and use CURL to test it. ```bash $ curl localhost/api?param1=test -i HTTP/1.1 200 OK Date: Sun, 08 Aug 2021 19:47:54 GMT Transfer-Encoding: chunked Server: Jetty(9.4.42.v20210604) ``` Now that we have a basic Request>Response flow, we want to add some logic in the middle. Let's quickly demonstrate returning a different response based on a URL parameter. Add a new **RouteOnAttribute** to the flow and enter the configuration. Add a new property with the **+** icon and name it **val1.** The value of this property will be: ``` ${http.param.param1:equals('val1')} ``` Do the same for **val2** as below. ![](./images/building-a-simple-rest-api-in-nifi-5.png) Create a copy of the HandleHTTPResponse (click on it, Ctrl+C, Ctrl+V) and modify the config of the copy. Change the **HTTP Status Code** to **201**. ![](./images/building-a-simple-rest-api-in-nifi-6.png) Now connect them all together. The **HandleHTTPRequest** should feed in to the **RouteOnAttribute**. From **RouteOnAttribute**, the **val1** relationship should go to the first **HandleHTTPResponse** with the 200 response code - the **val2** relationship goes to the second, with the 201 response code. See the screenshot below. ![](./images/building-a-simple-rest-api-in-nifi-7.png) Start all the processors and test it with cURL. In this cURL we are passing a URL parameter **param1** with the values **val1** or **val2**. This parameter becomes an attribute on the FlowFile for this request (_http.param.param1)_. The **RouteOnAttribute** accesses this attribute and routes based on the value, giving us a **200** response for **val1** or a **201** response for **val2**. ```bash $ curl localhost/api?param1=val1 -i HTTP/1.1 200 OK Date: Sun, 08 Aug 2021 19:52:05 GMT Transfer-Encoding: chunked Server: Jetty(9.4.42.v20210604) $ curl localhost/api?param1=val2 -i HTTP/1.1 201 Created Date: Sun, 08 Aug 2021 19:52:11 GMT Transfer-Encoding: chunked Server: Jetty(9.4.42.v20210604) ``` In reality, we don't really want to get different HTTP responses like this. The user of my API (the other NiFi flow) actually wanted some JSON in response. So let's do that. We want to return some JSON in the body of the resonse, so let's add a **ReplaceText**. As the FlowFile content is currently empty (null), we need to set **Replacement Strategy** to **Always Replace** and **Evaluation Mode** to **Entire Text** (you will see mime-type errors without these settings). Lastly, we need to set **Replacement Value** to the full content we want in our response, which is going to be a bit of JSON. ``` { "result": "you sent val1" } ``` ![](./images/building-a-simple-rest-api-in-nifi-8.png) We need to do the same thing for **val2**, so make a copy of the **ReplaceText** processor and change the JSON to say **val2** instead. ![](./images/building-a-simple-rest-api-in-nifi-9.png) We don't both **HandleHTTPResponse**'s anymore, so delete the second one. Now connect the **RouteOnAttribute** to the 2 **ReplaceText**'s and then connect the 2 **ReplaceText**'s to the single **HandleHTTPResponse**. ![](./images/building-a-simple-rest-api-in-nifi-2.png) Start all the processors and test it wth curl. ```bash $ curl localhost/api?param1=val1 { "result":"you sent val1" } $ curl localhost/api?param1=val2 { "result":"you sent val2" } ``` We now have a very basic REST API that is able to return different results based on user provided parameters. You can extend this with much more complexity - supporting more paths, allowing more HTTP methods, accepting more parameters, doing more complex routing logic and handling error with appropriate HTTP response codes. Find the flow definition [here](https://github.com/sdairs/nifi-snippets/tree/main/flows/simple-rest-api) if you'd like to import the whole thing. --- # Enriching Records with LookupRecord & REST APIs in NiFi Source: https://alasdairb.com/posts/enriching-records-with-lookuprecord-rest-apis-in-nifi Published: 2021-05-16 This is a pattern I have seen quite frequently, especially in IoT flows. At a high level: we have a stream of sensor data coming in from our IoT devices, and an external service that contains additional contextual data exposed via a REST API. With every sensor message, we want to call the REST API, passing some of the sensor information along, and recieve some additional related information in response. This isn't a particularly complex flow and it can be successfully implemented in several different ways. However, not all implementations are equal, so I'm going to demonstrate how to create this flow in a clean, efficient and maintainable way. Also, as I need a REST API to demonstrate the flow, I'll build a quick REST API inside NiFi to fulfill that, which I'll talk about [in another post](/posts/building-a-simple-rest-api-in-nifi). [Skip to solution](#solution) ## **To Record or not to Record?** The most common 'mistake' I see when implementing this flow is one that is quite prevelent in NiFi. That 'mistake' is not utilising NiFi's Record capabilities. This is a hugely powerful feature of NiFi that completely changes how a flow is implemented. It's more efficient, it's neater, it's easier to debug and easier to maintain. NiFi is designed to work with streams, and as such there does not need to be this 1:1 relationship of FlowFile to message. Instead, we can far more efficiently process a FlowFile that contains hundreds, even thousands, of messages at a time. We do this using Records. A Record can be any kind of message that can be defined by a Schema (even if you don't provide a Schema yourself) - a piece of JSON, some XML, a row in a CSV, an Avro message. None of this is specific to IoT, but it's a good example. Most sensors are sending out a series of small, consistent, well-formatted messages. If we only had 1 sensor sending a message every 5 seconds, it wouldn't really matter too much - but if we were to scale to multiple thousands of sensors, we end up having to deal with thousands of messages arriving at any one point in time, all the time. A continuous stream of messages. ## **Generating Test Data** I do not have a fleet of several thousands IoT devices, so I will be using **GenerateFlowFile** to produce messages instead. In the real world, you might be using MiNiFi to consume events at the edge and/or utilising some form of centralised message bus, like Kafka. I will use some Expression Langauge in the **GenerateFlowFile** to give us some random variance to messages so that we can do some routing logic in the API. My GenerateFlowFile settings are kept default, with the _Custom Text_ field set to: ```json { "someInt": "${random()}", "someUUID": "${UUID()}", "query": "${random():mod(2):equals(0):ifElse('val1','val2')}" } ``` Which gives me messages that look like this: ```json { "someInt": "3557167745430046240", "someUUID": "cc93197a-5393-470b-8057-c4ef79c811b8", "query": "val1" } ``` Where _someInt_ is a random number, _someUUID_ is a random UUID string and _query_ is one of either _val1_ or _val2_. Turning on this **GenerateFlowFlow** will give us individual FlowFiles with a single message in each. However, we want to simulate multiple message per FlowFile - imagine we are consuming from Kafka with a batch size of 1000. So let's merge some FlowFiles using a **MergeContent** processor. This will result in fewer FlowFiles with many messages in each. Note that we could also use a **MergeRecord** here, which would accurately reflect the output of say, a **ConsumeKafkaRecord** processor. I am keeping most settings at default, only changing _Minimum Number of Entries_ to 200 and the _Max Bin Age_ to 2 seconds. ![Configured MergeContent Picture](./images/enriching-records-with-lookuprecord-rest-apis-in-nifi-3.png) Configured MergeContent Now connect those 2 processors together and we are done with the traffic simulation. Remember, in reality you wouldn't need this bit, it's just for testing purposes. In the screenshot below you can see I have connected the _Success_ relationship from the **GenerateFlowFile** to the **MergeContent**. Then I have connected the _Merged_ relationship from the **MergeContent** to a funnel. The funnel is just so we can test this part of flow now and we will delete it later. I wrap the processors in a Label block to help make the flow easier to understand. ![GenerateFlowFile and MergeContent connected](./images/enriching-records-with-lookuprecord-rest-apis-in-nifi-1.png) GenerateFlowFile and MergeContent connected Start the flow and you'll see that a handful of FlowFiles build up in the _Success_ queue, before coming out as a single FlowFile in the _Merged_ queue. ![](./images/enriching-records-with-lookuprecord-rest-apis-in-nifi-12.png) List the _Merged_ queue and view the contents of one of the FlowFIles. You will see that we now have many similar messages in each FlowFile. (Right click the _Merged_ queue and click _List queue._ Next to one of the items in the list, click the little Eyeball icon.) ![](./images/enriching-records-with-lookuprecord-rest-apis-in-nifi-14.png) Perfect, we are finished generating our test data. ## The wrong way Most of the time when I see this flow being built, it ends up looking like the screenshot below and it actually makes a lot of sense. If you aren't too experienced with NiFi and you're just doing what seems logical, then you end up with this, and it does _work_. ![The wrong way to build this flow](./images/enriching-records-with-lookuprecord-rest-apis-in-nifi-15.png) The wrong way to build this flow That's one of the strengths of NiFi - it's very easy to build a flow that _works_. However, it's less than optimal and misses out on many of the other strengths of NiFi. Lets improve it. ## The right way You might be surprised to see that we can actually redo that entire flow with a single processor. Now, it's not quite _that_ simple, as there's some additional controller services that we need to configure and understand, but we end up with a much cleaner flow. The magic processor is **LookupRecord**. This processor allows us to take an input Record and perform a Lookup. We've already discussed Records, but what is a Lookup? Well, essentially it's performing a query and expecting a result. That sounds nice and generic, right? That's the beauty of the **LookupRecord** processor. There are so many different forms that a 'Lookup' might take. In our example here, its a REST API call - but what if you wanted to query an HBase table? Or an ElasticSearch index? Or a CSV file? Or execute a custom script? Well, **LookupRecord** encaspulates **[all of that and more](https://nifi.apache.org/docs/nifi-docs/components/org.apache.nifi/nifi-standard-nar/1.13.1/org.apache.nifi.processors.standard.LookupRecord/)!** It's a seriously powerful processor. Let's drag a **LookupRecord** into our canvas and take a look at its settings. First of all, we need Record Reader and Writer services. These services describe how NiFi should read and write our messages. In this example, we are only using JSON for our messages. NiFi has native support for understanding JSON, so let's add a new **JsonTreeReader** and **JsonRecordSetWriter** as our Reader and Writer. Click in to each field and from the drop down, select _Create New Service_. ![Adding a new JsonTreeReader service](./images/enriching-records-with-lookuprecord-rest-apis-in-nifi-16.png) Adding a new JsonTreeReader service For the Reader we are selecting **JsonTreeReader** and for the Writer we are selecting **JsonRecordSetWriter**_._ You should enter a descriptive name in the _Controller Service Name_ box so that you remember what it's used for. Create both services now, and we should have something like this. ![LookupRecord with Reader and Writer](./images/enriching-records-with-lookuprecord-rest-apis-in-nifi-17.png) LookupRecord with Reader and Writer We now need to configure & enable both the Reader and Writer. Click the right arrow next to one of them to get to the controller configuration window. ![The Controller Configuration window](./images/enriching-records-with-lookuprecord-rest-apis-in-nifi-2.png) The Controller Service configuration window with disabled services Next to each service, we can click the cog icon to review it's configuration. For this example, we are going to leave the configurations as default. In the real world, you can set the Schema for your data here, which is a great way to ensure confirmity and consistency of your data throughout your pipelines. To enable the services, click the lightning bolt icon and in the popup window click _Enable**.**_ Close the popup window when the services have finished enabling, and you'll see that the lightning bolt icon now has a line through it. If you need to change Controller Service configurations, you will need to disable them first by clicking the lighting bolt icon. ![](./images/enriching-records-with-lookuprecord-rest-apis-in-nifi-4.png) The Controller Service configuration window with enabled controllers With both services enabled, close the Configuration window and return to the **LookupRecord** config window. Now, add a _Lookup Service_ just as we did for the Record Reader/Writers. You'll see lots of options in the _Compatible Controller Services_ drop down, but we are looking for **Rest Lookup Service**. ![Adding a REST Lookup Service](./images/enriching-records-with-lookuprecord-rest-apis-in-nifi-5.png) Adding a REST Lookup Service Click _Create_ and then click the right arrow icon to go back to the Controller Service configuration window. Click the Cog icon to configure the _Rest Lookup Service_ controller. Firstly, we need to add a _Record Reader_ service just as we did before. This is the Reader that is used for understand the results of the API call. Create a new _JsonTreeReader_ with default settings & enable it (again you might want to use a Schema in the real world, but we aren't here). Secondly, we need to enter the URL to our REST API. As I mentioned at the start, I have created a dummy REST API within NiFi itself, so I am pointing back to my own NiFi instance. This needs to be a valid URL - if you use HTTPS you will need to configure an SSLContextService. My API is on the path _/api_ and I am passing a URL parameter called _param1_. ```json http://172.19.0.3/api?param1=${myQuery} ``` I am setting the value of _param1_ using Expression Language to a variable called _myQuery_. Now, you'll notice that we have never set a variable called _myQuery_ yet, but we do have a field in our Record called _query._ It is actually the value of the Record field _query_ that we want to use as the value of _param1._ We'll get to that soon - keep reading! We're finished configuring the _RestLookupService_, so click _Apply_ and then enable the service with the lighting bolt icon. ![Configured RestLookupService](./images/enriching-records-with-lookuprecord-rest-apis-in-nifi-6.png) Configured RestLookupService Return to the config window of the _LookupRecord_ processor. There's only 2 options left to change. We need to do something with the result of the API call we made. For this example, I just want to stick the JSON result in to the Record as a new field. I'm going to call that new field _enrich_. So I need to put the full _Record Path_ in to the _Result RecordPath_ value. Read more about Record Paths [here](https://nifi.apache.org/docs/nifi-docs/html/record-path-guide.html). I will use the value _/enrich_ - this means a field called _enrich_ at the root of the tree. ![Setting the Result RecordPath](./images/enriching-records-with-lookuprecord-rest-apis-in-nifi-7.png) Setting the Result RecordPath Lastly, we need to set that _myQuery_ variable to the value of our _query_ Record field. We do this by creating a new dynamic property in the _LookupRecord_ where the property name is the name of the variable (so _myQuery_) and the value is the _Record Path_ to the field in the Record (so _/query_). Add the new dynamic property by clicking the big **+** icon in the top right of the _LookupRecord_ configuration window. ![Finished LookupRecord config](./images/enriching-records-with-lookuprecord-rest-apis-in-nifi-8.png) Finished LookupRecord config Click Apply and we are finished with configuration. I will delete the _Funnel_ that my _MergeContent_ was connected to, and instead connect the _MergeContent_ directly in to the _LookupRecord_. I'll add 2 new _Funnels_ and connect the _failure_ and _success_ relationships from _LookupRecord_ to one _Funnel_ each. ![Complete flow](./images/enriching-records-with-lookuprecord-rest-apis-in-nifi-9.png) Complete flow The API that I created in NiFi in NiFi is very simple and I will discuss building it in a different post. At a high level, it is listening on _/api_ and looking for the _param1_ URL parameter. It checks to see whether _param1_ value is either _val1_ or _val2._ It returns a JSON response telling you which value was sent. E.g. for _val1_ it will send: ```json { "result": "you sent val1" } ``` ![](./images/enriching-records-with-lookuprecord-rest-apis-in-nifi-10.png) NiFi Dummy Rest API So, all that is left is to turn our flow on and see what we get. Right clicking the canvas background and clicking Start will start all processors in the _Process Group_ that we are in. Messages are being generated and then merged in to FlowFiles that contain multiple messages. These FlowFiles are read by the _LookupRecord_ which hits our REST API endpoint for _every Record_ within the FlowFile. The result returned by the API is inserted into each record in the _enrich_ field. ![Running Flow](./images/enriching-records-with-lookuprecord-rest-apis-in-nifi-11.png) Running Flow Let's check the results by inspecting the FlowFile content on the _success_ queue from the _LookupRecord_. ```json ...,{ "someInt" : "8421070184682044126", "someUUID" : "3db93282-16e2-4b0b-9c18-d8d822011872", "query" : "val2", "enrich" : { "result" : "you sent val2" } }, { "someInt" : "6962852573498787385", "someUUID" : "628b6989-44e4-43b3-b883-8d6fa34a7fa4", "query" : "val1", "enrich" : { "result" : "you sent val1" } },... ``` ![Inspecting the enriched data](./images/enriching-records-with-lookuprecord-rest-apis-in-nifi-13.png) Inspecting the enriched data As you can see, each of our Records now has an additiona field called _enrich_ which contains the JSON result from the REST API. ## Summary To summarise, we have built a flow that is able to use Record data as part of an API call and insert the API response in to a field of the same Record. We have done this using only a single processor with a couple of supporting Controller Services, improving upon a flow that might have used 5 or more Processors to achieve the same goal. You can download the Flow Definition for this flow [here](https://github.com/sdairs/nifi-snippets/tree/main/flows/enriching-records-lookuprecord). That's it! Thanks for reading. --- # Version Control & Deploy flows with NiFi Registry Source: https://alasdairb.com/posts/nifi-in-production-nifi-registry Published: 2021-03-22 In realworld deployments, we usually have more than one NiFi environment; one where we develop our flows (**Dev**), and one where we deploy and run the flows (**Prod**). It's quite common to see people building the flow in Dev, exporting it as a template in XML, then importing the template XML in to Prod. This works, but it's not ideal. There's a few drawbacks to this method: it's a completely new flow so we lose any saved state in stateful processors, it's a manual process, you need to clean up old versions, etc. Instead, we can use NiFi Registry to improve the process. NiFi Registry is like Git for your NiFi flows - in fact, you can add a Git provider to NiFi Registry which stores your flows in your Git host of choice (local Git, GitLab, GitHub, BitBucket, etc.) We can create a single NiFi Registry that is common between Dev and Prod, allowing us to build flows in Dev, commit them to Registry, and drag them in to the Prod canvas. We can then make changes to the flow in Dev, commit the changes to a new flow version in Registry, and finally pull changes down to the flow Prod. The versioning is baked in, there's no old flows to clean up, and we even retain state on stateful processors (e.g. ListFile). Let's walk through the process. This assume you have a working NiFi and NiFi Registry already installed, I am using NiFi 1.13.1 and Registry 0.8.0. If you don't, here's some helpful links: - [NiFi Registry Homepage](https://nifi.apache.org/projects/registry/) - [NiFi Registry Docs](https://nifi.apache.org/docs/nifi-registry-docs/index.html) Firstly, we see a blank registry with no buckets or flows. We click the Spanner icon in the top right. ![](./images/nifi-in-production-nifi-registry-3.png) We see an empty list of bucket. We click NEW BUCKET to add a bucket for our flows. ![](./images/nifi-in-production-nifi-registry-1.png) We get a diaglogue box that lets us name the bucket - for example, we can name it after the project that will store its flows here. ![](./images/nifi-in-production-nifi-registry-21.png) We can see the created bucket in our list. ![](./images/nifi-in-production-nifi-registry-22.png) Now, in our Dev NiFi, we need to connect to Registry. Go to the burger menu in the top right and click Controller Settings. ![](./images/nifi-in-production-nifi-registry-25.png) This opens a tabbed diaglogue, and we want the REGISTRY CLIENTS tab. Click the + icon to add a new client. ![](./images/nifi-in-production-nifi-registry-26.png) We get a dialogue box to enter the registry details. Add the URL for your NiFi Registry instance - just copy the URL from the browser, including the _http://_ prefix, hostname, port, and the _/nifi-registry/_ suffix. Provide a name that identifies your Registry client - you can have more than one NiFi Registry so this lets us indentify which Registry we are using when we version a flow. The description is optional. ![](./images/nifi-in-production-nifi-registry-2.png) We can see our NiFi Registry client in the list. ![](./images/nifi-in-production-nifi-registry-4.png) Go back to the Root canvas in NiFi and drag on a new Process Group. I will call mine FlowA. ![](./images/nifi-in-production-nifi-registry-23.png) Drag in some processors to create your flow. Mine is just a ListFile on _/tmp_ to demonstrate that the state is kept between versions. ![](./images/nifi-in-production-nifi-registry-24.png) When done with the flow, right click the canvas background, go to _Version_ and then _Start Version Control_. ![](./images/nifi-in-production-nifi-registry-5.png) We get a dialogue box to provide details. Select the Registy client and the Bucket we created earlier. Provide a name for the saved flow. A description can be given to describe the flow, and this is _not_ changed between versions. You can also provide Version Comments, similar to a commit message, which can be given with each new version. ![](./images/nifi-in-production-nifi-registry-6.png) Now, we can see that a Tick appears next to our Process Groups title, showing that version control is enabled and we are tracking the latest changes. ![](./images/nifi-in-production-nifi-registry-7.png) Back in NiFi Registry, we can see the flow is shown with various details. ![](./images/nifi-in-production-nifi-registry-8.png) Now, in our Prod NiFi, complete the same steps as above to add the Registry client. Once done, go to the Root canvas and drag in a new Process Group. This time, click the _Import_ link in the dialogue box. ![](./images/nifi-in-production-nifi-registry-9.png) We can pick the Registry client and Bucket we want to import flows from. Then we can pick the Flow to import. Then we are given a list of the available version to choose from. We only have 1 version - you need to click the row in the list. ![](./images/nifi-in-production-nifi-registry-10.png) Our flow is imported, and we see the tick just as we did in Dev. ![](./images/nifi-in-production-nifi-registry-11.png) Now, back in Dev NiFi, make some changes to the flow. I have added an UpdateAttribute processor. ![](./images/nifi-in-production-nifi-registry-12.png) Commit the changes to Registry by right clicking in the canvas background, then _Version_, then _Commit Local Changes_. ![](./images/nifi-in-production-nifi-registry-13.png) We get a dialogue box that lets us provide some comments on the version, and we see in the dark circle what the version number is. ![](./images/nifi-in-production-nifi-registry-14.png) Back in NiFi Registry, we can see that our flow has another version. ![](./images/nifi-in-production-nifi-registry-15.png) Back in Prod NiFi, the flow we imported from NiFi Registry now shows a red arrow icon, indicating there is a newer version of this flow available. ![](./images/nifi-in-production-nifi-registry-16.png) Inside the process group, we can see the flow is still the original version. To upgrade, we can right click the canvas background, go to _Version_ and then _Change Version._ ![](./images/nifi-in-production-nifi-registry-17.png) We see a dialogue box with a list of all the available flow version, their creation timestamp and the comments. Select the new version in the list. ![](./images/nifi-in-production-nifi-registry-18.png) A progress box will open. When you upgrade a flow to a new version, it will stop all processors in the flow. ![](./images/nifi-in-production-nifi-registry-19.png) We can now see that our flow has been updated in Prod. State for ListFile has been preserved and its ready to start. ![](./images/nifi-in-production-nifi-registry-20.png) This demonstrates how to utilise NiFi Registry to develop & deploy NiFi flows between your Dev and Prod NiFi environments; a much better process than relying on export/import of Flow templates. --- # NiFi Resources For Learning & Improving Source: https://alasdairb.com/posts/nifi-resources-for-learning-improving Published: 2020-11-23 It's always a bit of a learning curve to get started with any new tool, not to mention keeping up to date with a tool that is under heavy active development. Here's a list of invaluable resources to consult: **NiFi Anti-Pattners by Mark Payne** Mark Payne is the co-creator of NiFi and has a running series called 'NiFi Anti-Patterns'. It's a great place to start. - [Flow Structure](https://www.youtube.com/watch?v=RjWstt7nRVY) - [Organising Flows](https://www.youtube.com/watch?v=v1CoQk730qs) - [Load-Balancing](https://www.youtube.com/watch?v=by9P0Zi8Dk8) - [Scheduling](https://www.youtube.com/watch?v=pZq0EbfDBy4) **NiFi Docs** The homepage for the NiFi project is [here](https://nifi.apache.org/). There are several different guides maintained by the project: - [Docs Home](https://nifi.apache.org/components/) - [Getting Started Guide](https://nifi.apache.org/docs/nifi-docs/html/getting-started.html) - [Administration Guide](https://nifi.apache.org/docs/nifi-docs/html/administration-guide.html) - [User Guide](https://nifi.apache.org/docs/nifi-docs/html/user-guide.html) - [Record Path Guide](https://nifi.apache.org/docs/nifi-docs/html/record-path-guide.html) - [Expression Language Guide](https://nifi.apache.org/docs/nifi-docs/html/expression-language-guide.html) **Scripting cookbooks:** - [Cookbook Part 1](https://community.cloudera.com/t5/Community-Articles/ExecuteScript-Cookbook-part-1/ta-p/248922) - [Cookbook Part 2](https://community.cloudera.com/t5/Community-Articles/ExecuteScript-Cookbook-part-2/ta-p/249018) - [Cookbook Part 3](https://community.cloudera.com/t5/Community-Articles/ExecuteScript-Cookbook-part-3/ta-p/249148) **Blogs from the experts** [Blog from Tim Spann](https://www.datainmotion.dev/) [Blog from Pierre Villard](https://pierrevillard.com/) **On this blog** You can find a few posts I have written about NiFi: [NiFi Terminology Basics](/posts/nifi-terminology-basics) [S3 Event Notifications in NiFi](/posts/s3-event-notifications-in-nifi) [Importing individual flow definitions in NiFi](/posts/importing-individual-flow-definitions-in-nifi) [Enriching Records with LookupRecord & REST APIs in NiFi](/posts/enriching-records-with-lookuprecord-rest-apis-in-nifi) [Version Control & Deploy flows with NiFi Registry](/posts/nifi-in-production-nifi-registry) [Getting started with NiFi's ScriptedReader](/posts/getting-started-with-nifis-scriptedreader-by-processing-key-value-pairs-part-1) --- # Kafka with multiple Listeners and SASL Source: https://alasdairb.com/posts/kafka-with-multiple-listeners-and-sasl Published: 2020-11-07 This will quickly discuss how to configure multiple Listeners, with the intent of having a unique Listener for External/Client traffic and another for Internal/Inter-broker traffic (and how this can be done with Cloudera Manager which requires a slight work-around in the current versions pre-2021). There's several valid use cases for multiple Listeners. In this case, we have brokers with multiple NICs, where one NIC is designated for internal traffic, while the other NIC is for external traffic. **Example 1 - Vanilla Kafka, 2 NICs, different IPs, different Hostnames** Easy enough to do with Kafka on its own: ```bash listeners=PUBLIC://:9093, PRIVATE://:19093 listener.security.protocol.map=PUBLIC:SASL_SSL,PRIVATE:SASL_SSL advertised.listeners=PUBLIC_NIC://:9093, SASL_SSL://:19093 inter.broker.listener.name=PRIVATE ``` **Example 2 - Cloudera Manager + Kafka, 2 NICs, different IPs, different Hostnames** Unfortunately, there is a pending issue with Cloudera Manager right now that does not allow us to remove the security.inter.broker.protocol setting - which, annoyingly, means we can't use the same config as above - you'll see the following error: > org.apache.kafka.common.config.ConfigException: Only one of inter.broker.listener.name and security.inter.broker.protocol should be set. So we have a work around.... ```bash listeners=PUBLIC://:9093, SASL_SSL://:19093 listener.security.protocol.map=PUBLIC:SASL_SSL,SASL_SSL:SASL_SSL advertised.listeners=PUBLIC_NIC://:9093, SASL_SSL://:19093 ``` There are 2 things to note: - We have renamed the PRIVATE listener to SASL_SSL (same name as the protocol we want it to use) - We do not specify the inter.broker.listener.name We also need to ensure that the _Inter Broker Protocol_ setting in Cloudera Manager is set to match the Listener name for the private interface, i.e. SASL_SSL. We're basically tricking Cloudera Manager by naming the Listener exactly the same as its protocol. **Example 3 - Vanilla Kafka, 2 NICs, different IPs, same Hostnames** ```bash listeners=PUBLIC://:9093, PRIVATE://:19093 listener.security.protocol.map=PUBLIC:SASL_SSL,PRIVATE:SASL_SSL advertised.listeners=PUBLIC_NIC://:9093, SASL_SSL://:19093 inter.broker.listener.name=PRIVATE ``` **Example 4 - Cloudera Manager + Kafka, 2 NICs, different IPs, same Hostnames** Apply the same workaround as Example 2. ```bash listeners=PUBLIC://:9093, SASL_SSL://:19093 listener.security.protocol.map=PUBLIC:SASL_SSL,SASL_SSL:SASL_SSL advertised.listeners=PUBLIC_NIC://:9093, SASL_SSL://:19093 ``` --- # Connecting to a secure Impala via NiFi Source: https://alasdairb.com/posts/connecting-to-a-secure-impala-via-nifi-with-tls-and-kerberos Published: 2020-09-04 **Note: Please be aware that this JDBC driver is NOT fully supported by NiFi, thus there is no guarentee that more complex features/behavious will work properly (or at all). It's a good work around for simple use cases, but it should not be relied upon heavily.** There's a few different ways this could be done. I'll demonstrate one possible way, using ExecuteSQL to connect to Impala via the JDBC driver. This assumes both Kerberos and TLS are in use. We are also using an internal PKI, so we have to provide custom CACerts via a truststore. Get the Cloudera JDBC Driver from the Cloudera downloads page. Unzip it, and find the `_ImpalaJDBC4.jar_`. Move this jar to your NiFi extensions dir, by default in `/var/lib/nifi/extensions`. You need to put it on **all** NiFi nodes. Now, the file needs to have the appropriate permissions. ```bash chown nifi:nifi /var/lib/nifi/extensions/ImpalaJDBC4.jar chmod 770 /var/lib/nifi/extensions/ImpalaJDBC4.jar ``` Add an ExecuteSQL processor to your flow. In the drop down for the connection pool, add a new service. Select a DBConnectionPool service and give it a name. For the driver location, point it towards the new jar. ```bash /var/lib/nifi/extensions/ImpalaJDBC4.jar ``` For the driver class name, check the Cloudera Impala JDBC documentation for the available options. We'll use the following: ```bash com.cloudera.impala.jdbc4.Driver ``` Now for the connection string. This will depend heavily on your environment. In this case, I am using keytabs, so I will provide a KeytabCredentialService later on. ```bash jdbc:impala://:21050;AuthMech=1; KrbRealm=;KrbHostFQDN=; KrbServiceName=impala;SSL=1;SSLKeyStore=/keystore.jks;SSLKeyStorePwd=;SSLTrustStore=/truststore.jks ``` This is all that's needed for the DBConnectionPool. Save this and go back to the ExecuteSQL processor. In the Kerberos Credentials Service, add a new KeytabCredentialService. Provide a valid principal and keytab file - ensure the principal actually has permissions in Impala. Add a simple query in the select query field and give it a test. FYI: The same technique works for Hive2, which is handy because NiFi does not work with Hive2. An odd decision there. Anyway. Obtain the Hive JDBCs from the Cloudera downloads page. With some slight changes to the conection string, it's otherwise exactly the same. ```bash jdbc:hive2://:21050;AuthMech=1; KrbRealm=;KrbHostFQDN=; KrbServiceName=hive;SSL=1;SSLKeyStore=/keystore.jks;SSLKeyStorePwd=;SSLTrustStore=/truststore.jks ``` --- # Moving the data dir of MariaDB on CentOS7 Source: https://alasdairb.com/posts/moving-the-data-dir-of-mariadb-on-centos7-rhel7 Published: 2020-09-04 You should have root access to the CentOS host and a new target directory ready for the MariaDB data. For this guide, our new target directory is `/data/database`. First, stop MariaDB. `systemctl stop mariadb` Now, copy your existing database directory to the new location. By default, it is `/var/lib/mysql`. If it's not there, check the config in `/etc/my.cnf` and look for the datadir path. We use -r to copy recursively, and -p to preserve the permissions. You could also use rsync. `cp -rp /var/lib/mysql /data/database` Now rename your old datadir so we don't get confused. `mv /var/lib/mysql /var/lib/mysql.bak` Now edit the `/etc/my.cnf` file. We are changing any instance of the previous datadir of `/var/lib/mysql` to the new datadir of `/data/database/mysql`. `[mysqld] datadir = /data/database/mysql socket = /data/database/mysql/mysql.sock ... log_bin = /data/database/mysql/mysql_binary_log [client] port=3306 socket=/data/database/mysql/mysql.sock` Lastly, if you have binary logs on (which you will by default) you'll need to modify the index file. Edit the `mysql_binary_log.index` file, which will now be at `/data/database/mysql/mysql_binary_log.index`. Again, replace any instance of the previous datadir of `/var/lib/mysql` to the new datadir of `/data/database/mysql`. Example: `/var/lib/mysql/mysql_binary_log.000027` becomes `/data/database/mysql/mysql_binary_log.000027` Now start MariaDB. `systemctl start mariadb` Don't forget to clean up the `/var/lib/mysql.bak` after you have tested that your database is working.