Using BigQuery with Workers AI·续

Using BigQuery with Workers AI·续

return rowsWithFieldNames }

export default { async fetch(request, env, ctx) { ... // Transform output format into array of objects with named fields let formattedResults;

if ('rows' in ticketInfo) { formattedResults = formatRows(ticketInfo.rows, ticketInfo.schema.fields); console.log(formattedResults) } else if ('error' in ticketInfo) { return new Response(ticketInfo.error.message, { status: 500 }) } ... }, };


## 7. Feed data into Workers AI

Now that you have converted the response from the BigQuery API into an array of results, generate some tags and attach an associated sentiment score using an LLM via [Workers AI](https://developers.cloudflare.com/workers-ai/):

```javascript
const generateTags = (data, env) => {
    return env.AI.run("@cf/meta/llama-3.1-8b-instruct", {
        prompt: `Create three one-word tags for the following text. return only these three tags separated by a comma. don't return text that is not a category.Lowercase only. ${JSON.stringify(data)}`,
    });
}

const generateSentimentScore = (data, env) => {
    return env.AI.run("@cf/meta/llama-3.1-8b-instruct", {
        prompt: `return a float number between 0 and 1 measuring the sentiment of the following text. 0 being negative and 1 positive. return only the number, no text. ${JSON.stringify(data)}`,
    });
}

// Iterates through values, sends them to an AI handler and encapsulates all responses into a single Promise
const getAIGeneratedContent = (data, env, aiHandler) => {
    let results = data?.map(dataPoint => {
        return aiHandler(dataPoint, env)
    })
    return Promise.all(results)
}
...
export default {
    async fetch(request, env, ctx) {
        ...
let summaries, sentimentScores;
        try {
            summaries = await getAIGeneratedContent(formattedResults, env, generateTags);
            sentimentScores = await getAIGeneratedContent(formattedResults, env, generateSentimentScore)
        } catch {
            return new Response('There was an error while generating the text summaries or sentiment scores')
        }
},

formattedResults = formattedResults?.map((formattedResult, i) => {
            if (sentimentScores[i].response && summaries[i].response) {
                return {
                    ...formattedResult,
                    'sentiment': parseFloat(sentimentScores[i].response).toFixed(2),
                    'tags': summaries[i].response.split(',').map((result) => result.trim())
                }
            }
        }
};

Uncomment the following lines from the Wrangler file in your project:

{
    "ai": {
        "binding": "AI"
    }
}

[ai]
binding = "AI"

Restart the Worker that is running locally, and after doing so, go to your application endpoint:

curl http://localhost:8787

It is likely that you will be asked to log in to your Cloudflare account and grant temporary access to Wrangler (the Cloudflare CLI) to use your account when using Worker AI.

Once you access http://localhost:8787 you should see an output similar to the following:

{
  "data": [
    {
    "text": "You can see a clear spike in submissions right around US Thanksgiving.",
    "sentiment": "0.61",
    "tags": [
        "trends",
        "submissions",
        "thanksgiving"
    ]
    },
    {
    "text": "I didn't test the changes before I published them.  I basically did development on the running server. In fact for about 30 seconds the comments page was broken due to a bug.",
    "sentiment": "0.35",
    "tags": [
        "software",
        "deployment",
        "error"
    ]
    },
    {
    "text": "I second that. As I recall, it's a very enjoyable 700-page brain dump by someone who's really into his subject. The writing has a personal voice; there are lots of asides, dry wit, and typos that suggest restrained editing. The discussion is intelligent and often theoretical (and Bartle is not scared to use mathematical metaphors), but the tone is not academic.",
    "sentiment": "0.86",
    "tags": [
        "review",
        "game",
        "design"
    ]
    }
  ]
}

The actual values and fields will mostly depend on the query made in Step 5 that is then fed into the LLM.

Final result

All the code shown in the different steps is combined into the following code in src/index.js:

import * as jose from "jose";

const generateBQJWT = async (env) => {
    const algorithm = "RS256";
    const audience = "https://bigquery.googleapis.com/";
    const expiryAt = new Date().valueOf() / 1000;
    const privateKey = await jose.importPKCS8(env.BQ_PRIVATE_KEY, algorithm);

// Generate signed JSON Web Token (JWT)
    return new jose.SignJWT()
        .setProtectedHeader({
            typ: "JWT",
            alg: algorithm,
            kid: env.BQ_PRIVATE_KEY_ID,
        })
        .setIssuer(env.BQ_CLIENT_EMAIL)
        .setSubject(env.BQ_CLIENT_EMAIL)
        .setAudience(audience)
        .setExpirationTime(expiryAt)
        .setIssuedAt()
        .sign(privateKey);
};

const queryBQ = async (bgJWT, path) => {
    const bqEndpoint = `https://bigquery.googleapis.com${path}`;
    const query = "SELECT text FROM hn.news_sampled LIMIT 3";
    const response = await fetch(bqEndpoint, {
        method: "POST",
        body: JSON.stringify({
            query: query,
        }),
        headers: {
            Authorization: `Bearer ${bgJWT}`,
        },
    });
    return response.json();
};

const formatRows = (rowsWithoutFieldNames, fields) => {
    // Index to fieldName
    const fieldsByIndex = new Map();

fields.forEach((field, index) => {
        fieldsByIndex.set(index, field.name);
    });

const rowsWithFieldNames = rowsWithoutFieldNames.map((row) => {
        // Map rows into an array of objects with field names
        let newRow = {};
        row.f.forEach((field, index) => {
            const fieldName = fieldsByIndex.get(index);
            if (fieldName) {
                newRow = { ...newRow, [fieldName]: field.v };
            }
        });
        return newRow;
    });

return rowsWithFieldNames;
};

const generateTags = (data, env) => {
    return env.AI.run("@cf/meta/llama-3.1-8b-instruct", {
        prompt: `Create three one-word tags for the following text. return only these three tags separated by a comma. don't return text that is not a category.Lowercase only. ${JSON.stringify(data)}`,
    });
};

const generateSentimentScore = (data, env) => {
    return env.AI.run("@cf/meta/llama-3.1-8b-instruct", {
        prompt: `return a float number between 0 and 1 measuring the sentiment of the following text. 0 being negative and 1 positive. return only the number, no text. ${JSON.stringify(data)}`,
    });
};

const getAIGeneratedContent = (data, env, aiHandler) => {
    let results = data?.map((dataPoint) => {
        return aiHandler(dataPoint, env);
    });
    return Promise.all(results);
};

export default {
    async fetch(request, env, ctx) {
        // Create JWT to authenticate the BigQuery API call
        let bqJWT;
        try {
            bqJWT = await generateBQJWT(env);
        } catch (error) {
            console.log(error);
            return new Response("An error has occurred while generating the JWT", {
                status: 500,
            });
        }

// Fetch results from BigQuery
        let ticketInfo;
        try {
            ticketInfo = await queryBQ(
                bqJWT,
                `/bigquery/v2/projects/${env.BQ_PROJECT_ID}/queries`,
            );
        } catch (error) {
            console.log(error);
            return new Response("An error has occurred while querying BQ", {
                status: 500,
            });
        }

// Transform output format into array of objects with named fields
        let formattedResults;
        if ("rows" in ticketInfo) {
            formattedResults = formatRows(ticketInfo.rows, ticketInfo.schema.fields);
        } else if ("error" in ticketInfo) {
            return new Response(ticketInfo.error.message, { status: 500 });
        }

// Generate AI summaries and sentiment scores
        let summaries, sentimentScores;
        try {
            summaries = await getAIGeneratedContent(
                formattedResults,
                env,
                generateTags,
            );
            sentimentScores = await getAIGeneratedContent(
                formattedResults,
                env,
                generateSentimentScore,
            );
        } catch {
            return new Response(
                "There was an error while generating the text summaries or sentiment scores",
            );
        }

// Add AI summaries and sentiment scores to previous results
        formattedResults = formattedResults?.map((formattedResult, i) => {
            if (sentimentScores[i].response && summaries[i].response) {
                return {
                    ...formattedResult,
                    sentiment: parseFloat(sentimentScores[i].response).toFixed(2),
                    tags: summaries[i].response.split(",").map((result) => result.trim()),
                };
            }
        });

const response = { data: formattedResults };

return new Response(JSON.stringify(response), {
            headers: { "Content-Type": "application/json" },
        });
    },
};

If you wish to deploy this Worker, you can do so by running npx wrangler deploy:

Total Upload: &lt;size_of_your_worker> KiB / gzip: &lt;compressed_size_of_your_worker> KiB
Uploaded &lt;name_of_your_worker> (x sec)
Deployed &lt;name_of_your_worker> triggers (x sec)
  https://&lt;your_public_worker_endpoint>
Current Version ID: &lt;worker_script_version_id>

This will create a public endpoint that you can use to access the Worker globally. Please keep this in mind when using production data, and make sure to include additional access controls in place.

Conclusion

In this tutorial, you have learnt how to integrate Google BigQuery and Cloudflare Workers by creating a GCP service account key and storing part of it as Worker secrets. This was later imported in the code, and by using the jose npm library, you created a JSON Web Token to authenticate the API query to BigQuery.

Once you obtained the results, you formatted them to pass to generative AI models via Workers AI to generate tags and to perform sentiment analysis on the extracted data.

Next Steps

If, instead of displaying the results of ingesting the data to the AI model in a browser, your workflow requires fetching and store data (for example in R2 or D1) on regular intervals, you may want to consider adding a scheduled handler for this Worker. This enables you to trigger the Worker with a predefined cadence via a Cron Trigger. Consider reviewing the Reference Architecture Diagrams on Ingesting BigQuery Data into Workers AI.

A use case to ingest data from other sources, like you did in this tutorial, is to create a RAG system. If this sounds relevant to you, please check out the Build a Retrieval Augmented Generation (RAG) AI tutorial.

To learn more about what other AI models you can use at Cloudflare, please visit the Workers AI section of our docs.

Was this helpful?

YesNo

{"@context":"https://schema.org","@type":"TechArticle","@id":"https://developers.cloudflare.com/workers-ai/guides/tutorials/using-bigquery-with-workers-ai/#page","headline":"Using BigQuery with Workers AI","description":"Learn how to ingest data stored outside of Cloudflare as an input to Workers AI models.","url":"https://developers.cloudflare.com/workers-ai/guides/tutorials/using-bigquery-with-workers-ai/","inLanguage":"en","image":"https://developers.cloudflare.com/workers-ai/guides/tutorials/using-bigquery-with-workers-ai/og.png?v=3ee0e03210117266","dateModified":"2026-01-29","publisher":{"@type":"Organization","name":"Cloudflare","description":"One platform for your apps, agents, and workforce. Build, secure, and scale without managing infrastructure","url":"https://www.cloudflare.com/","sameAs":["https://github.com/cloudflare","https://www.linkedin.com/company/cloudflare","https://x.com/cloudflare"],"logo":{"@type":"ImageObject","url":"https://developers.cloudflare.com/logo.svg"},"address":{"@type":"PostalAddress","streetAddress":"101 Townsend St","addressLocality":"San Francisco","addressRegion":"CA","postalCode":"94107","addressCountry":"US"},"contactPoint":[{"@type":"ContactPoint","contactType":"Customer Support","url":"https://support.cloudflare.com/","availableLanguage":["English"]},{"@type":"ContactPoint","contactType":"Sales","url":"https://www.cloudflare.com/contact/","availableLanguage":["English"]}]},"isPartOf":{"@type":"WebSite","@id":"https://developers.cloudflare.com/#website","name":"Cloudflare Docs","url":"https://developers.cloudflare.com/"},"keywords":["AI","JavaScript"]}

description: Observe and control your AI applications with analytics, caching, rate limiting, and model fallback through AI Gateway. title: Cloudflare AI Gateway image: https://developers.cloudflare.com/ai-gateway/og.png?v=d9b4c0d9dbab4986

Documentation Index
Fetch the complete documentation index at: https://developers.cloudflare.com/ai-gateway/llms.txt
Use this file to discover all available pages before exploring further.

Using BigQuery with Workers AI·续

Using BigQuery with Workers AI·续

Final result

Conclusion

Next Steps

On this page