@t = EXTRACT 
date string,
time string,
author string,
tweet string
FROM "/input/MyTwitterHistory.csv" USING Extractors.Csv();

@res = 
SELECT author,
COUNT(\*) AS tweetcount
FROM @t GROUP BY author;

OUTPUT @res TO "/output/MyTwitterAnalysis.csv"ORDER BY tweetcount DESC
USING Outputters.Csv();

using Microsoft.Analytics.Interfaces;
using Microsoft.Analytics.Interfaces.Streaming;
using Microsoft.Analytics.Types.Sql;
using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Text;
using Newtonsoft.Json;
using Newtonsoft.Json.Linq;
using Utilities;

namespace Utilities
{
    //Use Atomic if we want to operate on the whole file vs line by line
    [SqlUserDefinedExtractor(AtomicFileProcessing = false)]
    public class CustomTextExtractor : IExtractor
    {
        //Contains the row
        private readonly Encoding _encoding;
        private readonly byte[] _row_delim;
        private readonly char _col_delim;

        public CustomTextExtractor(Encoding encoding = null, string row_delim = "\r\n", char col_delim = '|')
        {
            _encoding = Encoding.UTF8;
            _row_delim = _encoding.GetBytes(row_delim);
            _col_delim = col_delim;
        }

        public override IEnumerable<IRow> Extract(IUnstructuredReader input, IUpdatableRow output)
        {

            //Read the input line by line
            foreach (Stream current in input.Split(_row_delim))
            {
                using (StreamReader streamReader = new StreamReader(current, _encoding))
                {
                    string[] parts = streamReader.ReadToEnd()
                        .Trim().Split(new char[] { _col_delim }, StringSplitOptions.None);

                    int count = 0; // start with first column
                    foreach (string part in parts)
                    {
                        if (count == 3)
                        {
                            string json = part;
                            var log = JsonConvert.DeserializeObject<RootObject>(json);
                            var items = new SqlArray<string>(new string[] 
                            {
                                log.header.appName,
                                log.header.currentContext,
                                log.header.documentID,
                                log.header.endTime,
                                log.header.errorCode,
                                log.header.message,
                                log.header.rootContext,
                                log.header.server,
                                log.header.startTime,
                            });

                            output.Set<SqlArray<string>>(count, items);
                            count += 1;
                        }
                        else
                        {
                            //    // keep the rest of the columns as-is
                            output.Set(count, part);
                            count += 1;
                        }

                    }
                }
                yield return output.AsReadOnly();
            }
            yield break;
        }

    }

}

// A. CREATE ASSEMBLY: Register assemblies (if they do not already exist).
CREATE ASSEMBLY IF NOT EXISTS [Newtonsoft.Json] FROM @"adl://alandatalakeadls.azuredatalakestore.net/Newtonsoft.Json.dll";
CREATE ASSEMBLY IF NOT EXISTS [Microsoft.Analytics.Samples.Formats] FROM  @"adl://alandatalakeadls.azuredatalakestore.net/Microsoft.Analytics.Samples.Formats.dll";
DROP ASSEMBLY IF EXISTS [Utilities];
CREATE ASSEMBLY IF NOT EXISTS [Utilities] FROM @"adl://alandatalakeadls.azuredatalakestore.net/Utilities.dll";
// B. REFERENCE ASSEMBLY: Load assemblies for compile time and execution.
REFERENCE ASSEMBLY [Newtonsoft.Json];
REFERENCE ASSEMBLY [Microsoft.Analytics.Samples.Formats];
REFERENCE ASSEMBLY [Utilities];

// C. USING: Specify namespace to shorten function names (e.g. Microsoft.Analytics.Samples.Formats.Json.JsonExtractor)
USING Microsoft.Analytics.Samples.Formats.Json;

// 1. Initialise variables for Input (e.g. JSON) and Output (e.g. CSV).
DECLARE @InputFile string = @"adl://alandatalakeadls.azuredatalakestore.net/BreakDownOfCode.txt";
DECLARE @OutputFile string = @"adl://alandatalakeadls.azuredatalakestore.net/output.csv";

@RawExtract  = EXTRACT 
        Date string,              
        Status  string,           
        PriceNotification  string,
        header SqlArray <string>
FROM @InputFile 
USING new Utilities.CustomTextExtractor(col_delim:'|',row_delim:"\n");

//Change the data in some way if we need to such as if an item is null coalesce to an empty string. Call C# methods.
@IntermediaryData =
SELECT
    Date,
    Status,
    PriceNotification,
    header
FROM @RawExtract;

@OutputData =
SELECT
    Date,
    Status,
    PriceNotification,
    header[0] AS appName,
    header[1] AS currentContext,
    header[2] AS documentID,
    header[3] AS endTime,
    header[4] AS errorCode,
    header[5] AS message,
    header[6] AS rootContext,
    header[7] AS server,
    header[8] AS startTime 
FROM @IntermediaryData;

OUTPUT @OutputData
TO @OutputFile
USING Outputters.Csv(outputHeader:true, quoting : false);