flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From JP de Vooght ...@vooght.de>
Subject Re: logging question
Date Sat, 07 Apr 2018 12:46:45 GMT
Nico, all,

I am still stuck with this. Upgraded the docker image to 1.4.2 and the
AMIDST library to 0.7.0

Just noticed this issue which signals logging issues outside transforms:
https://issues.apache.org/jira/browse/FLINK-7990

Could this be related? Although I don't see the relation to logback.

Below is the library code invoked after "BEFORE updateModel"

|try { Configuration config = new Configuration();
config.setString(BN_NAME, this.dag.getName());
config.setBytes(EFBN_NAME,
Serialization.serializeObject(efBayesianNetwork)); DataSet<DataInstance>
dataset = dataUpdate.getDataSet(); this.sumSS = dataset.map(new
SufficientSatisticsMAP()) .withParameters(config) .reduce(new
SufficientSatisticsReduce()) .collect().get(0); //Add the prior
sumSS.sum(efBayesianNetwork.createInitSufficientStatistics());
JobExecutionResult result =
dataset.getExecutionEnvironment().getLastJobExecutionResult();
numInstances =
result.getAccumulatorResult(ParallelMaximumLikelihood.COUNTER_NAME+"_"+this.dag.getName());
numInstances++;//Initial counts }catch(Exception ex){ throw new
UndeclaredThrowableException(ex); } |

JP

On 01/16/2018 10:50 AM, Nico Kruber wrote:
> Just a guess, but probably our logging initialisation changes the global
> log level (see conf/log4j.properties). DataStream.collect() executes the
> program along with creating a local Flink "cluster" (if you are testing
> locally / in an IDE) and initializing logging, among other things.
>
> Please comment the first line out and uncomment the following one to
> read like this:
> ==========
> # This affects logging for both user code and Flink
> #log4j.rootLogger=INFO, file
>
> # Uncomment this if you want to _only_ change Flink's logging
> log4j.logger.org.apache.flink=INFO
> ==========
>
>
> Nico
>
> On 13/01/18 13:52, jp@vooght.de wrote:
>> Hello,
>> I am learning Flink and using the docker image along with the AMIDST
>> library for this.
>> Below is a sample task from AMIDST which provides INFO output up until I
>> reach updateModel(). I pasted the short method as well and wonder what
>> prevents the Logger from
>>
>>         //Set-up Flink session
>>         env = ExecutionEnvironment.getExecutionEnvironment();
>>         env.getConfig().disableSysoutLogging();
>>         Logger LOG = LoggerFactory.getLogger(">>>>> ParallelMLExample");
>>
>>         //generate a random dataset
>>         DataFlink<DataInstance> dataFlink = new
>> DataSetGenerator().generate(env, 1234, 1000, 5, 0);
>>
>>         //Creates a DAG with the NaiveBayes structure for the random
>> dataset
>>         DAG dag =
>> DAGGenerator.getNaiveBayesStructure(dataFlink.getAttributes(),
>> "DiscreteVar4");
>>         LOG.info(dag.toString());
>>
>>         //Create the Learner object
>>         ParameterLearningAlgorithm learningAlgorithmFlink = new
>> ParallelMaximumLikelihood();
>>
>>         //Learning parameters
>>         learningAlgorithmFlink.setBatchSize(10);
>>         learningAlgorithmFlink.setDAG(dag);
>>
>>         //Initialize the learning process
>>         learningAlgorithmFlink.initLearning();
>>
>>         //Learn from the flink data
>>         LOG.info("BEFORE UPDATEMODEL");
>>         learningAlgorithmFlink.updateModel(dataFlink);
>>         LOG.info("AFTER UPDATEMODEL");
>>
>>         //Print the learnt Bayes Net
>>         BayesianNetwork bn =
>> learningAlgorithmFlink.getLearntBayesianNetwork();
>>         LOG.info(bn.toString());
>>
>>
>> Below is the updateModel method.
>>
>>     public double updateModel(DataFlink<DataInstance> dataUpdate) {
>>         try {
>>             Configuration config = new Configuration();
>>             config.setString(BN_NAME, this.dag.getName());
>>             config.setBytes(EFBN_NAME,
>> Serialization.serializeObject(efBayesianNetwork));
>>
>>             DataSet<DataInstance> dataset = dataUpdate.getDataSet();
>>             this.sumSS = dataset.map(new SufficientSatisticsMAP())
>>                     .withParameters(config)
>>                     .reduce(new SufficientSatisticsReduce())
>>                     .collect().get(0);
>>
>>             //Add the prior
>>             sumSS.sum(efBayesianNetwork.createInitSufficientStatistics());
>>
>>             JobExecutionResult result =
>> dataset.getExecutionEnvironment().getLastJobExecutionResult();
>>
>>             numInstances =
>> result.getAccumulatorResult(ParallelMaximumLikelihood.COUNTER_NAME+"_"+this.dag.getName());
>>
>>             numInstances++;//Initial counts
>>
>>         }catch(Exception ex){
>>             throw new UndeclaredThrowableException(ex);
>>         }
>>
>>         return this.getLogMarginalProbability();
>>     }
>>
>>
>> Not sure why LOG.info past that method are not output to the console.
>> TIA
>> JP


Mime
View raw message